bottleneckcrm/backend/tasks/tests/test_tasks_api.py
Ashwin dd6ca2958d Add unit tests for organization admin checks and saved views provider
- Implement tests for `isOrgAdminProvider` to validate admin roles based on organization membership and role.
- Create tests for `saved_views_provider` to ensure correct API interactions for saving, renaming, and deleting saved views.
- Enhance ticket tree tests to accurately reflect inactive and merged ticket handling.
- Update CSV import tests to verify error handling and file saving functionality.
- Introduce tests for deal cards to ensure correct display of next steps and task statuses.
2026-09-26 20:38:32 +05:30

2122 lines
73 KiB
Python

import json
import pytest
from django.contrib.contenttypes.models import ContentType
from django.core.files.uploadedfile import SimpleUploadedFile
from accounts.models import Account
from cases.models import Case
from common.models import Attachments, Comment, Tags, Teams
from contacts.models import Contact
from leads.models import Lead
from opportunity.models import Opportunity
from tasks.models import Task
@pytest.mark.django_db
class TestTaskListView:
"""Tests for GET /api/tasks/ and POST /api/tasks/."""
def test_list_tasks(self, admin_client, admin_user, org_a):
Task.objects.create(
title="Sample Task",
status="New",
priority="Medium",
org=org_a,
created_by=admin_user,
)
response = admin_client.get("/api/tasks/")
assert response.status_code == 200
data = response.json()
assert "tasks" in data
assert data["tasks_count"] >= 1
def test_create_task(self, admin_client, org_a):
response = admin_client.post(
"/api/tasks/",
{"title": "New Task", "status": "New", "priority": "High"},
format="json",
)
assert response.status_code == 200
data = response.json()
assert data["error"] is False
assert Task.objects.filter(title="New Task", org=org_a).exists()
def test_create_task_unauthenticated(self, unauthenticated_client):
"""An anonymous POST is refused with a 403.
This used to assert that `PermissionDenied` propagated out of the
client call, which it never does: DRF's exception handler turns it
into a response, which is the whole point of raising it. The endpoint
was behaving correctly the entire time the test was red, so assert the
refusal the caller actually sees, and that nothing was written.
"""
response = unauthenticated_client.post(
"/api/tasks/",
{"title": "Unauthorized Task", "status": "New", "priority": "Low"},
format="json",
)
assert response.status_code == 403
assert not Task.objects.filter(title="Unauthorized Task").exists()
def test_org_isolation(self, admin_client, admin_user, org_a, org_b):
"""Tasks from another org should not appear in the task list."""
Task.objects.create(
title="Org A Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
# Create a task in org_b that should not be visible
from common.models import User
other_user = User.objects.create_user(
email="other@example.com", password="testpass123"
)
Task.objects.create(
title="Org B Task",
status="New",
priority="Low",
org=org_b,
created_by=other_user,
)
response = admin_client.get("/api/tasks/")
assert response.status_code == 200
data = response.json()
titles = [t["title"] for t in data["tasks"]]
assert "Org A Task" in titles
assert "Org B Task" not in titles
def test_create_task_with_all_optional_fields(
self, admin_client, admin_profile, admin_user, org_a
):
"""Create a task with contacts, teams, assigned_to, tags, and due_date."""
contact = Contact.objects.create(
first_name="John",
last_name="Doe",
org=org_a,
created_by=admin_user,
)
team = Teams.objects.create(name="Dev Team", org=org_a, created_by=admin_user)
tag = Tags.objects.create(name="urgent", slug="urgent", org=org_a)
response = admin_client.post(
"/api/tasks/",
{
"title": "Full Task",
"status": "New",
"priority": "High",
"due_date": "2026-12-31",
"description": "A detailed description",
"contacts": [str(contact.id)],
"teams": [str(team.id)],
"assigned_to": [str(admin_profile.id)],
"tags": [str(tag.id)],
},
format="json",
)
assert response.status_code == 200
assert response.json()["error"] is False
task = Task.objects.get(title="Full Task")
assert task.contacts.count() == 1
assert task.teams.count() == 1
assert task.assigned_to.count() == 1
assert task.tags.count() == 1
assert str(task.due_date) == "2026-12-31"
def test_two_tasks_may_share_a_title(self, admin_client, admin_user, org_a):
"""Titles are not unique per org, and should not be.
This test used to assert a 400 for a duplicate title. No such
constraint exists on `Task`, and adding one would be wrong: "Call
back", "Send the quote" and "Chase invoice" are the titles a real task
list is *made* of, one per customer. The rule the test wanted belongs
to a ticket subject, not to a to-do.
"""
Task.objects.create(
title="Call back",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.post(
"/api/tasks/",
{"title": "Call back", "status": "New", "priority": "Low"},
format="json",
)
assert response.status_code == 200
assert Task.objects.filter(title="Call back", org=org_a).count() == 2
def test_list_tasks_filter_by_status(self, admin_client, admin_user, org_a):
"""Filtering tasks by status query param."""
Task.objects.create(
title="New Task A",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
Task.objects.create(
title="Completed Task B",
status="Completed",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.get("/api/tasks/?status=New")
assert response.status_code == 200
data = response.json()
titles = [t["title"] for t in data["tasks"]]
assert "New Task A" in titles
assert "Completed Task B" not in titles
def test_list_tasks_filter_by_priority(self, admin_client, admin_user, org_a):
"""Filtering tasks by priority query param."""
Task.objects.create(
title="High Prio",
status="New",
priority="High",
org=org_a,
created_by=admin_user,
)
Task.objects.create(
title="Low Prio",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.get("/api/tasks/?priority=High")
assert response.status_code == 200
data = response.json()
titles = [t["title"] for t in data["tasks"]]
assert "High Prio" in titles
assert "Low Prio" not in titles
def test_list_tasks_filter_by_title(self, admin_client, admin_user, org_a):
"""Filtering tasks by title query param."""
Task.objects.create(
title="Alpha Work",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
Task.objects.create(
title="Beta Work",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.get("/api/tasks/?title=Alpha")
assert response.status_code == 200
data = response.json()
titles = [t["title"] for t in data["tasks"]]
assert "Alpha Work" in titles
assert "Beta Work" not in titles
def test_list_tasks_filter_by_search(self, admin_client, admin_user, org_a):
"""Search query param should filter by title."""
Task.objects.create(
title="Searchable Item",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
Task.objects.create(
title="Hidden Item",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.get("/api/tasks/?search=Searchable")
assert response.status_code == 200
data = response.json()
assert data["tasks_count"] == 1
assert data["tasks"][0]["title"] == "Searchable Item"
def test_list_tasks_filter_by_assigned_to(
self, admin_client, admin_profile, admin_user, org_a
):
"""Filtering tasks by assigned_to query param."""
task = Task.objects.create(
title="Assigned Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
task.assigned_to.add(admin_profile)
Task.objects.create(
title="Unassigned Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.get(f"/api/tasks/?assigned_to={admin_profile.id}")
assert response.status_code == 200
data = response.json()
titles = [t["title"] for t in data["tasks"]]
assert "Assigned Task" in titles
assert "Unassigned Task" not in titles
def test_list_tasks_filter_by_tags(self, admin_client, admin_user, org_a):
"""Filtering tasks by tags query param."""
tag = Tags.objects.create(name="backend", slug="backend", org=org_a)
task = Task.objects.create(
title="Tagged Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
task.tags.add(tag)
Task.objects.create(
title="Untagged Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.get(f"/api/tasks/?tags={tag.id}")
assert response.status_code == 200
data = response.json()
titles = [t["title"] for t in data["tasks"]]
assert "Tagged Task" in titles
assert "Untagged Task" not in titles
def test_list_tasks_filter_by_due_date_range(self, admin_client, admin_user, org_a):
"""Filtering tasks by due_date__gte and due_date__lte."""
Task.objects.create(
title="Jan Task",
status="New",
priority="Low",
due_date="2026-01-15",
org=org_a,
created_by=admin_user,
)
Task.objects.create(
title="Dec Task",
status="New",
priority="Low",
due_date="2026-12-15",
org=org_a,
created_by=admin_user,
)
response = admin_client.get(
"/api/tasks/?due_date__gte=2026-06-01&due_date__lte=2026-12-31"
)
assert response.status_code == 200
data = response.json()
titles = [t["title"] for t in data["tasks"]]
assert "Dec Task" in titles
assert "Jan Task" not in titles
def test_list_tasks_non_admin_sees_only_assigned(
self, user_client, regular_user, admin_user, org_a, user_profile
):
"""A non-admin user should only see tasks assigned to them.
Note: ORM-created tasks have created_by=None (crum middleware not active),
so only assigned_to filtering works in tests.
"""
Task.objects.create(
title="Admin Only Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
assigned_task = Task.objects.create(
title="User Assigned Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
assigned_task.assigned_to.add(user_profile)
response = user_client.get("/api/tasks/")
assert response.status_code == 200
data = response.json()
titles = [t["title"] for t in data["tasks"]]
assert "User Assigned Task" in titles
assert "Admin Only Task" not in titles
def test_list_tasks_response_includes_metadata(
self, admin_client, admin_user, org_a
):
"""Response should include status, priority, accounts_list, contacts_list.
The convention `cases` and `opportunity` already follow, the list
endpoint hands the form the pickers it needs instead of making the
client fetch three more catalogues. Tasks did not follow it, so this
test was red rather than wrong.
"""
Task.objects.create(
title="Meta Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.get("/api/tasks/")
assert response.status_code == 200
data = response.json()
assert "status" in data
assert "priority" in data
assert "accounts_list" in data
assert "contacts_list" in data
@pytest.mark.django_db
class TestTaskDetailView:
"""Tests for GET/PUT/DELETE /api/tasks/<pk>/."""
def test_get_detail(self, admin_client, admin_user, org_a):
task = Task.objects.create(
title="Detail Task",
status="New",
priority="Medium",
org=org_a,
created_by=admin_user,
)
response = admin_client.get(f"/api/tasks/{task.id}/")
assert response.status_code == 200
data = response.json()
assert "task_obj" in data
assert data["task_obj"]["title"] == "Detail Task"
def test_update_task(self, admin_client, admin_user, org_a):
task = Task.objects.create(
title="Old Title",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.put(
f"/api/tasks/{task.id}/",
{"title": "Updated Title", "status": "In Progress", "priority": "High"},
format="json",
)
assert response.status_code == 200
data = response.json()
assert data["error"] is False
task.refresh_from_db()
assert task.title == "Updated Title"
assert task.status == "In Progress"
def test_delete_task(self, admin_client, admin_user, org_a):
task = Task.objects.create(
title="Delete Me",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.delete(f"/api/tasks/{task.id}/")
assert response.status_code == 200
data = response.json()
assert data["error"] is False
assert not Task.objects.filter(id=task.id).exists()
def test_cross_org(self, org_b_client, admin_user, org_a):
"""A client from org_b must not access a task belonging to org_a."""
task = Task.objects.create(
title="Org A Only",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = org_b_client.get(f"/api/tasks/{task.id}/")
assert response.status_code == 404
def test_add_comment(self, admin_client, admin_user, org_a):
task = Task.objects.create(
title="Comment Task",
status="New",
priority="Medium",
org=org_a,
created_by=admin_user,
)
response = admin_client.post(
f"/api/tasks/{task.id}/",
{"comment": "This is a test comment"},
format="json",
)
assert response.status_code == 200
data = response.json()
assert "comments" in data
def test_create_task_invalid(self, admin_client):
"""POST without a title should return 400."""
response = admin_client.post(
"/api/tasks/",
{"status": "New", "priority": "Low"},
format="json",
)
assert response.status_code == 400
data = response.json()
assert data["error"] is True
def test_update_task_with_tags_and_assigned_to(
self, admin_client, admin_profile, admin_user, org_a
):
"""PUT should clear and re-set tags and assigned_to."""
tag1 = Tags.objects.create(name="tag1", slug="tag1", org=org_a)
tag2 = Tags.objects.create(name="tag2", slug="tag2", org=org_a)
task = Task.objects.create(
title="Tag Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
task.tags.add(tag1)
task.assigned_to.add(admin_profile)
# Update: replace tag1 with tag2, keep assigned_to
response = admin_client.put(
f"/api/tasks/{task.id}/",
{
"title": "Tag Task",
"status": "New",
"priority": "Low",
"tags": [str(tag2.id)],
"assigned_to": [str(admin_profile.id)],
},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
tag_ids = list(task.tags.values_list("id", flat=True))
assert tag2.id in tag_ids
assert tag1.id not in tag_ids
assert task.assigned_to.count() == 1
def test_update_task_clears_m2m_when_empty(
self, admin_client, admin_profile, admin_user, org_a
):
"""PUT without contacts/teams/assigned_to/tags should clear them."""
contact = Contact.objects.create(
first_name="Jane",
last_name="Doe",
org=org_a,
created_by=admin_user,
)
task = Task.objects.create(
title="Clear M2M Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
task.contacts.add(contact)
task.assigned_to.add(admin_profile)
response = admin_client.put(
f"/api/tasks/{task.id}/",
{
"title": "Clear M2M Task",
"status": "New",
"priority": "Low",
},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.contacts.count() == 0
assert task.assigned_to.count() == 0
assert task.tags.count() == 0
assert task.teams.count() == 0
def test_update_task_with_contacts_and_teams(self, admin_client, admin_user, org_a):
"""PUT should set contacts and teams."""
contact = Contact.objects.create(
first_name="Alice",
last_name="Smith",
org=org_a,
created_by=admin_user,
)
team = Teams.objects.create(name="QA Team", org=org_a, created_by=admin_user)
task = Task.objects.create(
title="Team Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.put(
f"/api/tasks/{task.id}/",
{
"title": "Team Task",
"status": "New",
"priority": "Low",
"contacts": [str(contact.id)],
"teams": [str(team.id)],
},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.contacts.count() == 1
assert task.teams.count() == 1
def test_update_task_invalid_returns_400(self, admin_client, admin_user, org_a):
"""PUT with invalid data returns 400."""
task = Task.objects.create(
title="Invalid Update Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
# Missing required title
response = admin_client.put(
f"/api/tasks/{task.id}/",
{"status": "New", "priority": "Low"},
format="json",
)
assert response.status_code == 400
def test_patch_task_partial_update(self, admin_client, admin_user, org_a):
"""PATCH should allow partial updates without requiring all fields."""
task = Task.objects.create(
title="Patch Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.patch(
f"/api/tasks/{task.id}/",
{"status": "In Progress"},
format="json",
)
assert response.status_code == 200
assert response.json()["error"] is False
task.refresh_from_db()
assert task.status == "In Progress"
assert task.title == "Patch Task" # Title unchanged
def test_patch_task_with_m2m_fields(
self, admin_client, admin_profile, admin_user, org_a
):
"""PATCH should handle M2M fields when present in request."""
tag = Tags.objects.create(name="patchtag", slug="patchtag", org=org_a)
task = Task.objects.create(
title="Patch M2M Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.patch(
f"/api/tasks/{task.id}/",
{
"tags": [str(tag.id)],
"assigned_to": [str(admin_profile.id)],
},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.tags.count() == 1
assert task.assigned_to.count() == 1
def test_patch_task_clear_m2m(self, admin_client, admin_profile, admin_user, org_a):
"""PATCH with empty list for M2M fields should clear them."""
task = Task.objects.create(
title="Patch Clear Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
task.assigned_to.add(admin_profile)
assert task.assigned_to.count() == 1
response = admin_client.patch(
f"/api/tasks/{task.id}/",
{"assigned_to": [], "tags": []},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.assigned_to.count() == 0
assert task.tags.count() == 0
def test_patch_non_admin_permission_denied(self, user_client, admin_user, org_a):
"""Non-admin who is not creator/assignee should get 403 on PATCH."""
task = Task.objects.create(
title="Admin Created Patch Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = user_client.patch(
f"/api/tasks/{task.id}/",
{"status": "In Progress"},
format="json",
)
assert response.status_code == 403
def test_patch_non_admin_as_assignee_via_patch(
self, user_client, admin_user, user_profile, org_a
):
"""Non-admin who is assigned to the task should be able to PATCH.
Note: The task view's permission check compares request.profile (Profile)
with task.created_by (User), which never matches. So creator-based
permission doesn't work in the task view. Only assigned_to works for
non-admin users.
"""
task = Task.objects.create(
title="User Patch Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
task.assigned_to.add(user_profile)
response = user_client.patch(
f"/api/tasks/{task.id}/",
{"status": "In Progress"},
format="json",
)
assert response.status_code == 200
assert response.json()["error"] is False
def test_patch_non_admin_as_assignee_allowed(
self, user_client, admin_user, user_profile, org_a
):
"""Non-admin who is assigned should be able to PATCH."""
task = Task.objects.create(
title="Assignee Patch Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
task.assigned_to.add(user_profile)
response = user_client.patch(
f"/api/tasks/{task.id}/",
{"status": "Completed"},
format="json",
)
assert response.status_code == 200
assert response.json()["error"] is False
def test_get_detail_response_fields(self, admin_client, admin_user, org_a):
"""GET detail should include all expected fields."""
task = Task.objects.create(
title="Fields Task",
status="New",
priority="Medium",
org=org_a,
created_by=admin_user,
)
response = admin_client.get(f"/api/tasks/{task.id}/")
assert response.status_code == 200
data = response.json()
assert "task_obj" in data
assert "attachments" in data
assert "comments" in data
assert "users_mention" in data
assert "assigned_data" in data
assert "users" in data
assert "teams" in data
def test_get_detail_non_admin_as_assignee(
self, admin_client, user_client, user_profile, org_a
):
"""A non-admin assignee can open the task they were handed.
This used to assert the opposite: `pytest.raises(AttributeError,
match="has no attribute 'user'")`, because the view read
`created_by.user.email` when `created_by` is already the `User`. The
test was green and the endpoint was unusable for every non-admin in
the product. Asserting the crash is not coverage; it is a bug with a
test holding it in place.
"""
create_response = admin_client.post(
"/api/tasks/",
{
"title": "User Detail Task",
"status": "New",
"priority": "Low",
"assigned_to": [str(user_profile.id)],
},
format="json",
)
assert create_response.status_code == 200
task = Task.objects.get(title="User Detail Task")
response = user_client.get(f"/api/tasks/{task.id}/")
assert response.status_code == 200
assert response.json()["task_obj"]["title"] == "User Detail Task"
def test_get_detail_non_admin_forbidden(self, user_client, admin_user, org_a):
"""Somebody on neither side of the task is refused with a 403.
Previously `pytest.raises(TypeError, match="not JSON serializable")`:
`get_context_data` returned a `Response`, `get()` wrapped it in a
second one, and the only branch meant to say "no" was a 500.
"""
task = Task.objects.create(
title="Forbidden Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = user_client.get(f"/api/tasks/{task.id}/")
assert response.status_code == 403
def test_delete_non_admin_not_the_creator_forbidden(
self, user_client, admin_user, org_a
):
"""A member who neither made nor was handed the task cannot delete it.
The assertion was always right; the docstring used to explain it by
the `Profile`/`User` comparison bug, which made every creator a
non-creator. That bug is fixed, and this member really is not the
creator; `created_by` below is the admin.
"""
task = Task.objects.create(
title="User Delete Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = user_client.delete(f"/api/tasks/{task.id}/")
assert response.status_code == 403
# Task should still exist
assert Task.objects.filter(id=task.id).exists()
def test_delete_non_admin_not_creator_forbidden(
self, user_client, admin_user, org_a
):
"""Non-admin who is not creator should get 403 on delete."""
task = Task.objects.create(
title="Admin Delete Only Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = user_client.delete(f"/api/tasks/{task.id}/")
assert response.status_code == 403
def test_post_comment_non_admin_forbidden(self, user_client, admin_user, org_a):
"""Non-admin who is not creator/assignee should get 403 on POST (comment)."""
task = Task.objects.create(
title="Comment Forbidden Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = user_client.post(
f"/api/tasks/{task.id}/",
{"comment": "Should fail"},
format="json",
)
assert response.status_code == 403
def test_post_comment_non_admin_as_assignee(
self, user_client, admin_user, user_profile, org_a
):
"""Non-admin who is assigned should be able to post a comment."""
task = Task.objects.create(
title="Assignee Comment Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
task.assigned_to.add(user_profile)
response = user_client.post(
f"/api/tasks/{task.id}/",
{"comment": "Assignee comment"},
format="json",
)
assert response.status_code == 200
assert "comments" in response.json()
@pytest.mark.django_db
class TestTaskCommentView:
"""Tests for PUT/PATCH/DELETE /api/tasks/comment/<pk>/."""
def _create_task_with_comment(self, admin_user, admin_profile, org_a):
"""Helper to create a task and a comment on it."""
task = Task.objects.create(
title="Comment Test Task",
status="New",
priority="Medium",
org=org_a,
created_by=admin_user,
)
ct = ContentType.objects.get_for_model(Task)
comment = Comment.objects.create(
content_type=ct,
object_id=task.id,
comment="Original comment",
commented_by=admin_profile,
org=org_a,
)
return task, comment
def test_update_comment_put_edits_the_text(
self, admin_client, admin_user, admin_profile, org_a
):
"""PUT with only `comment` edits the comment.
This used to assert a 400 and describe it as correct: the serializer
ran non-partial with `object_id` and `org` as required fields, so an
ordinary "fix my typo" edit failed validation. Those two are the
server's to set, are now read-only, and no longer stand between an
author and their own text.
"""
_task, comment = self._create_task_with_comment(
admin_user, admin_profile, org_a
)
response = admin_client.put(
f"/api/tasks/comment/{comment.id}/",
{"comment": "Updated comment text"},
format="json",
)
assert response.status_code == 200
comment.refresh_from_db()
assert comment.comment == "Updated comment text"
def test_update_comment_patch(self, admin_client, admin_user, admin_profile, org_a):
"""Admin should be able to partially update a comment via PATCH."""
_task, comment = self._create_task_with_comment(
admin_user, admin_profile, org_a
)
response = admin_client.patch(
f"/api/tasks/comment/{comment.id}/",
{"comment": "Patched comment"},
format="json",
)
assert response.status_code == 200
assert response.json()["error"] is False
assert response.json()["message"] == "Comment Updated"
def test_delete_comment(self, admin_client, admin_user, admin_profile, org_a):
"""Admin should be able to delete a comment."""
_task, comment = self._create_task_with_comment(
admin_user, admin_profile, org_a
)
response = admin_client.delete(f"/api/tasks/comment/{comment.id}/")
assert response.status_code == 200
assert response.json()["error"] is False
assert not Comment.objects.filter(id=comment.id).exists()
def test_update_comment_non_admin_forbidden(
self, user_client, user_profile, admin_user, admin_profile, org_a
):
"""Non-admin who did not create the comment should get 403."""
task, comment = self._create_task_with_comment(admin_user, admin_profile, org_a)
# A reader of the task: the author rule, not the read rule, refuses.
# One who cannot open the task gets the 404 a missing comment gets.
task.assigned_to.add(user_profile)
response = user_client.put(
f"/api/tasks/comment/{comment.id}/",
{"comment": "Should fail"},
format="json",
)
assert response.status_code == 403
def test_delete_comment_non_admin_forbidden(
self, user_client, user_profile, admin_user, admin_profile, org_a
):
"""Non-admin who did not create the comment should get 403 on delete."""
task, comment = self._create_task_with_comment(admin_user, admin_profile, org_a)
# A reader of the task: the author rule, not the read rule, refuses.
# One who cannot open the task gets the 404 a missing comment gets.
task.assigned_to.add(user_profile)
response = user_client.delete(f"/api/tasks/comment/{comment.id}/")
assert response.status_code == 403
def test_put_comment_without_comment_field_is_400_not_403(
self, admin_client, admin_user, admin_profile, org_a
):
"""A missing `comment` is a bad request, not a permission problem.
This asserted 403 and the comment underneath ("falls through") named
the defect without treating it as one: `if params.get("comment")` sat
inside the authorization branch, so an empty body skipped the
serializer and landed on the 403, telling the author they may not edit
their own comment.
"""
_task, comment = self._create_task_with_comment(
admin_user, admin_profile, org_a
)
response = admin_client.put(
f"/api/tasks/comment/{comment.id}/",
{},
format="json",
)
assert response.status_code == 400
assert "comment" in response.json()["errors"]
def test_put_blank_comment_is_400(
self, admin_client, admin_user, admin_profile, org_a
):
"""The other half of the same guard: `""` is falsy too."""
_task, comment = self._create_task_with_comment(
admin_user, admin_profile, org_a
)
original = comment.comment
response = admin_client.put(
f"/api/tasks/comment/{comment.id}/",
{"comment": ""},
format="json",
)
assert response.status_code == 400
comment.refresh_from_db()
assert comment.comment == original
def test_stranger_still_gets_403_for_an_empty_body(
self, user_client, user_profile, admin_user, admin_profile, org_a
):
"""Authorization is still checked before the body is."""
task, comment = self._create_task_with_comment(admin_user, admin_profile, org_a)
# A reader of the task: the author rule, not the read rule, refuses.
# One who cannot open the task gets the 404 a missing comment gets.
task.assigned_to.add(user_profile)
response = user_client.put(
f"/api/tasks/comment/{comment.id}/",
{},
format="json",
)
assert response.status_code == 403
@pytest.mark.django_db
class TestTaskAttachmentView:
"""Tests for DELETE /api/tasks/attachment/<pk>/."""
def test_delete_attachment(self, admin_client, admin_user, org_a):
"""Admin should be able to delete an attachment."""
from django.core.files.uploadedfile import SimpleUploadedFile
task = Task.objects.create(
title="Attachment Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
ct = ContentType.objects.get_for_model(Task)
attachment = Attachments.objects.create(
content_type=ct,
object_id=task.id,
file_name="test.txt",
attachment=SimpleUploadedFile("test.txt", b"file content"),
created_by=admin_user,
org=org_a,
)
response = admin_client.delete(f"/api/tasks/attachment/{attachment.id}/")
assert response.status_code == 200
assert response.json()["error"] is False
assert not Attachments.objects.filter(id=attachment.id).exists()
def test_delete_attachment_non_admin_forbidden(
self, user_client, user_profile, admin_user, org_a
):
"""Non-admin who did not create the attachment should get 403."""
from django.core.files.uploadedfile import SimpleUploadedFile
task = Task.objects.create(
title="Att Forbidden Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
ct = ContentType.objects.get_for_model(Task)
attachment = Attachments.objects.create(
content_type=ct,
object_id=task.id,
file_name="test.txt",
attachment=SimpleUploadedFile("test.txt", b"file content"),
created_by=admin_user,
org=org_a,
)
# A reader of the task who did not upload the file.
task.assigned_to.add(user_profile)
response = user_client.delete(f"/api/tasks/attachment/{attachment.id}/")
assert response.status_code == 403
# ---------------------------------------------------------------------------
# Additional coverage tests for uncovered lines in task_views.py
# ---------------------------------------------------------------------------
@pytest.mark.django_db
class TestTaskListFiltersCoverage:
"""Tests covering filter parameters in TaskListView.get_context_data
that were previously uncovered (lines 86-101).
"""
def test_filter_by_created_at_gte(self, admin_client, admin_user, org_a):
"""Filter tasks by created_at__gte."""
Task.objects.create(
title="Created At GTE Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.get("/api/tasks/?created_at__gte=2020-01-01")
assert response.status_code == 200
titles = [t["title"] for t in response.json()["tasks"]]
assert "Created At GTE Task" in titles
def test_filter_by_created_at_lte(self, admin_client, admin_user, org_a):
"""Filter tasks by created_at__lte."""
Task.objects.create(
title="Created At LTE Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.get("/api/tasks/?created_at__lte=2099-12-31")
assert response.status_code == 200
titles = [t["title"] for t in response.json()["tasks"]]
assert "Created At LTE Task" in titles
def test_filter_by_created_at_range(self, admin_client, admin_user, org_a):
"""Filter tasks by created_at__gte and created_at__lte together."""
Task.objects.create(
title="Range Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.get(
"/api/tasks/?created_at__gte=2020-01-01&created_at__lte=2099-12-31"
)
assert response.status_code == 200
assert response.json()["tasks_count"] >= 1
def test_filter_by_account(self, admin_client, admin_user, org_a):
"""Filter tasks by account FK."""
account = Account.objects.create(
name="Test Account Filter", org=org_a, created_by=admin_user
)
Task.objects.create(
title="Account Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
account=account,
)
Task.objects.create(
title="No Account Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.get(f"/api/tasks/?account={account.id}")
assert response.status_code == 200
titles = [t["title"] for t in response.json()["tasks"]]
assert "Account Task" in titles
assert "No Account Task" not in titles
def test_filter_by_opportunity(self, admin_client, admin_user, org_a):
"""Filter tasks by opportunity FK."""
opp = Opportunity.objects.create(
name="Test Opp Filter",
stage="PROSPECTING",
org=org_a,
created_by=admin_user,
)
Task.objects.create(
title="Opp Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
opportunity=opp,
)
Task.objects.create(
title="No Opp Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.get(f"/api/tasks/?opportunity={opp.id}")
assert response.status_code == 200
titles = [t["title"] for t in response.json()["tasks"]]
assert "Opp Task" in titles
assert "No Opp Task" not in titles
def test_filter_by_case(self, admin_client, admin_user, org_a):
"""Filter tasks by case FK."""
case = Case.objects.create(
name="Test Case Filter",
status="New",
priority="High",
org=org_a,
created_by=admin_user,
)
Task.objects.create(
title="Case Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
case=case,
)
Task.objects.create(
title="No Case Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.get(f"/api/tasks/?case={case.id}")
assert response.status_code == 200
titles = [t["title"] for t in response.json()["tasks"]]
assert "Case Task" in titles
assert "No Case Task" not in titles
def test_filter_by_lead(self, admin_client, admin_user, org_a):
"""Filter tasks by lead FK."""
lead = Lead.objects.create(
first_name="Test",
last_name="Lead Filter",
org=org_a,
created_by=admin_user,
)
Task.objects.create(
title="Lead Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
lead=lead,
)
Task.objects.create(
title="No Lead Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.get(f"/api/tasks/?lead={lead.id}")
assert response.status_code == 200
titles = [t["title"] for t in response.json()["tasks"]]
assert "Lead Task" in titles
assert "No Lead Task" not in titles
@pytest.mark.django_db
class TestTaskCreateAssociations:
"""Tests for task creation with FK associations (opportunity, case, lead, account)
covering lines 228-250 in task_views.py.
"""
def test_create_task_with_opportunity(self, admin_client, admin_user, org_a):
"""Create a task linked to an opportunity."""
opp = Opportunity.objects.create(
name="Create Opp Link",
stage="PROSPECTING",
org=org_a,
created_by=admin_user,
)
response = admin_client.post(
"/api/tasks/",
{
"title": "Opp Linked Task",
"status": "New",
"priority": "High",
"opportunity": str(opp.id),
},
format="json",
)
assert response.status_code == 200
assert response.json()["error"] is False
task = Task.objects.get(title="Opp Linked Task")
assert task.opportunity_id == opp.id
def test_create_task_with_case(self, admin_client, admin_user, org_a):
"""Create a task linked to a case."""
case = Case.objects.create(
name="Create Case Link",
status="New",
priority="High",
org=org_a,
created_by=admin_user,
)
response = admin_client.post(
"/api/tasks/",
{
"title": "Case Linked Task",
"status": "New",
"priority": "High",
"case": str(case.id),
},
format="json",
)
assert response.status_code == 200
assert response.json()["error"] is False
task = Task.objects.get(title="Case Linked Task")
assert task.case_id == case.id
def test_create_task_with_lead(self, admin_client, admin_user, org_a):
"""Create a task linked to a lead."""
lead = Lead.objects.create(
first_name="Create",
last_name="Lead Link",
org=org_a,
created_by=admin_user,
)
response = admin_client.post(
"/api/tasks/",
{
"title": "Lead Linked Task",
"status": "New",
"priority": "High",
"lead": str(lead.id),
},
format="json",
)
assert response.status_code == 200
assert response.json()["error"] is False
task = Task.objects.get(title="Lead Linked Task")
assert task.lead_id == lead.id
def test_create_task_with_contacts_as_json_string(
self, admin_client, admin_user, org_a
):
"""Create a task with contacts passed as a JSON string (line 176)."""
contact = Contact.objects.create(
first_name="JSON",
last_name="Contact",
org=org_a,
created_by=admin_user,
)
response = admin_client.post(
"/api/tasks/",
{
"title": "JSON Contact Task",
"status": "New",
"priority": "Low",
"contacts": json.dumps([str(contact.id)]),
},
format="json",
)
assert response.status_code == 200
task = Task.objects.get(title="JSON Contact Task")
assert task.contacts.count() == 1
def test_create_task_with_assigned_to_as_json_string(
self, admin_client, admin_profile, org_a
):
"""Create a task with assigned_to as JSON string (line 202)."""
response = admin_client.post(
"/api/tasks/",
{
"title": "JSON Assigned Task",
"status": "New",
"priority": "Low",
"assigned_to": json.dumps([str(admin_profile.id)]),
},
format="json",
)
assert response.status_code == 200
task = Task.objects.get(title="JSON Assigned Task")
assert task.assigned_to.count() == 1
def test_create_task_with_teams_as_json_string(
self, admin_client, admin_user, org_a
):
"""Create a task with teams as JSON string (line 190)."""
team = Teams.objects.create(name="JSON Team", org=org_a, created_by=admin_user)
response = admin_client.post(
"/api/tasks/",
{
"title": "JSON Team Task",
"status": "New",
"priority": "Low",
"teams": json.dumps([str(team.id)]),
},
format="json",
)
assert response.status_code == 200
task = Task.objects.get(title="JSON Team Task")
assert task.teams.count() == 1
def test_create_task_with_tags_as_json_string(
self, admin_client, admin_user, org_a
):
"""Create a task with tags as JSON string (line 216)."""
tag = Tags.objects.create(name="jsontag", slug="jsontag", org=org_a)
response = admin_client.post(
"/api/tasks/",
{
"title": "JSON Tag Task",
"status": "New",
"priority": "Low",
"tags": json.dumps([str(tag.id)]),
},
format="json",
)
assert response.status_code == 200
task = Task.objects.get(title="JSON Tag Task")
assert task.tags.count() == 1
@pytest.mark.django_db
class TestTaskDetailCreatedByCoverage:
"""Tests for task detail view covering the created_by == request.profile path
(lines 274-275, 309, 315, 319-320) in get_context_data.
"""
def test_detail_admin_as_creator(self, admin_client, admin_profile, org_a):
"""Admin who created the task via API sees full detail."""
create_resp = admin_client.post(
"/api/tasks/",
{"title": "Creator Detail Task", "status": "New", "priority": "Low"},
format="json",
)
assert create_resp.status_code == 200
task = Task.objects.get(title="Creator Detail Task")
response = admin_client.get(f"/api/tasks/{task.id}/")
assert response.status_code == 200
data = response.json()
assert "task_obj" in data
assert "users_mention" in data
assert "users_excluding_team" in data
def test_detail_non_admin_users_section(
self, admin_client, user_client, user_profile, admin_user, org_a
):
"""A non-admin assignee gets `users_mention` and a narrowed `users`.
Someone who did not create the task can mention its creator, and is
offered the org's admins to reassign to, not the whole org. This
also used to assert `AttributeError`: the mention list was built from
`created_by.user.email`, and `created_by` is the `User`.
"""
create_resp = admin_client.post(
"/api/tasks/",
{
"title": "Non Admin Detail Task",
"status": "New",
"priority": "Low",
"assigned_to": [str(user_profile.id)],
},
format="json",
)
assert create_resp.status_code == 200
task = Task.objects.get(title="Non Admin Detail Task")
body = user_client.get(f"/api/tasks/{task.id}/").json()
assert body["users_mention"] == [{"username": admin_user.email}]
assert [u["user_details"]["email"] for u in body["users"]] == [admin_user.email]
@pytest.mark.django_db
class TestTaskUpdateAssociations:
"""Tests for task PUT update with FK associations (opportunity, case, lead)
covering lines 530-552 in task_views.py.
"""
def test_update_task_with_opportunity(self, admin_client, admin_user, org_a):
"""PUT update: set an opportunity on a task."""
opp = Opportunity.objects.create(
name="Update Opp",
stage="PROSPECTING",
org=org_a,
created_by=admin_user,
)
task = Task.objects.create(
title="Update Opp Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.put(
f"/api/tasks/{task.id}/",
{
"title": "Update Opp Task",
"status": "New",
"priority": "Low",
"opportunity": str(opp.id),
},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.opportunity_id == opp.id
def test_update_task_clear_opportunity(self, admin_client, admin_user, org_a):
"""PUT update: clear opportunity by passing empty value in params."""
opp = Opportunity.objects.create(
name="Clear Opp",
stage="PROSPECTING",
org=org_a,
created_by=admin_user,
)
task = Task.objects.create(
title="Clear Opp Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
opportunity=opp,
)
response = admin_client.put(
f"/api/tasks/{task.id}/",
{
"title": "Clear Opp Task",
"status": "New",
"priority": "Low",
"opportunity": "",
},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.opportunity_id is None
def test_update_task_with_case(self, admin_client, admin_user, org_a):
"""PUT update: set a case on a task."""
case = Case.objects.create(
name="Update Case Link",
status="New",
priority="High",
org=org_a,
created_by=admin_user,
)
task = Task.objects.create(
title="Update Case Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.put(
f"/api/tasks/{task.id}/",
{
"title": "Update Case Task",
"status": "New",
"priority": "Low",
"case": str(case.id),
},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.case_id == case.id
def test_update_task_clear_case(self, admin_client, admin_user, org_a):
"""PUT update: clear case by passing empty value."""
case = Case.objects.create(
name="Clear Case Link",
status="New",
priority="High",
org=org_a,
created_by=admin_user,
)
task = Task.objects.create(
title="Clear Case Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
case=case,
)
response = admin_client.put(
f"/api/tasks/{task.id}/",
{
"title": "Clear Case Task",
"status": "New",
"priority": "Low",
"case": "",
},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.case_id is None
def test_update_task_with_lead(self, admin_client, admin_user, org_a):
"""PUT update: set a lead on a task."""
lead = Lead.objects.create(
first_name="Update",
last_name="Lead Link",
org=org_a,
created_by=admin_user,
)
task = Task.objects.create(
title="Update Lead Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.put(
f"/api/tasks/{task.id}/",
{
"title": "Update Lead Task",
"status": "New",
"priority": "Low",
"lead": str(lead.id),
},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.lead_id == lead.id
def test_update_task_clear_lead(self, admin_client, admin_user, org_a):
"""PUT update: clear lead by passing empty value."""
lead = Lead.objects.create(
first_name="Clear",
last_name="Lead Link",
org=org_a,
created_by=admin_user,
)
task = Task.objects.create(
title="Clear Lead Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
lead=lead,
)
response = admin_client.put(
f"/api/tasks/{task.id}/",
{
"title": "Clear Lead Task",
"status": "New",
"priority": "Low",
"lead": "",
},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.lead_id is None
def test_update_task_with_contacts_as_json_string(
self, admin_client, admin_user, org_a
):
"""PUT update with contacts as JSON string (covers line 475)."""
contact = Contact.objects.create(
first_name="UpdJSON",
last_name="Contact",
org=org_a,
created_by=admin_user,
)
task = Task.objects.create(
title="UpdJSON Contact Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.put(
f"/api/tasks/{task.id}/",
{
"title": "UpdJSON Contact Task",
"status": "New",
"priority": "Low",
"contacts": json.dumps([str(contact.id)]),
},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.contacts.count() == 1
def test_update_task_with_teams_as_json_string(
self, admin_client, admin_user, org_a
):
"""PUT update with teams as JSON string (covers line 490)."""
team = Teams.objects.create(
name="UpdJSON Team", org=org_a, created_by=admin_user
)
task = Task.objects.create(
title="UpdJSON Team Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.put(
f"/api/tasks/{task.id}/",
{
"title": "UpdJSON Team Task",
"status": "New",
"priority": "Low",
"teams": json.dumps([str(team.id)]),
},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.teams.count() == 1
def test_update_task_with_assigned_to_as_json_string(
self, admin_client, admin_profile, org_a
):
"""PUT update with assigned_to as JSON string (covers line 503)."""
task = Task.objects.create(
title="UpdJSON Assigned Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_profile.user,
)
response = admin_client.put(
f"/api/tasks/{task.id}/",
{
"title": "UpdJSON Assigned Task",
"status": "New",
"priority": "Low",
"assigned_to": json.dumps([str(admin_profile.id)]),
},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.assigned_to.count() == 1
def test_update_task_with_tags_as_json_string(
self, admin_client, admin_user, org_a
):
"""PUT update with tags as JSON string (covers line 518)."""
tag = Tags.objects.create(name="updjsontag", slug="updjsontag", org=org_a)
task = Task.objects.create(
title="UpdJSON Tag Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.put(
f"/api/tasks/{task.id}/",
{
"title": "UpdJSON Tag Task",
"status": "New",
"priority": "Low",
"tags": json.dumps([str(tag.id)]),
},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.tags.count() == 1
@pytest.mark.django_db
class TestTaskPatchAssociations:
"""Tests for task PATCH with FK associations (opportunity, case, lead)
covering lines 672-705 in task_views.py.
"""
def test_patch_set_opportunity(self, admin_client, admin_user, org_a):
"""PATCH: set opportunity on a task."""
opp = Opportunity.objects.create(
name="Patch Opp",
stage="PROSPECTING",
org=org_a,
created_by=admin_user,
)
task = Task.objects.create(
title="Patch Opp Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.patch(
f"/api/tasks/{task.id}/",
{"opportunity": str(opp.id)},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.opportunity_id == opp.id
def test_patch_clear_opportunity(self, admin_client, admin_user, org_a):
"""PATCH: clear opportunity by passing null/empty."""
opp = Opportunity.objects.create(
name="Patch Clear Opp",
stage="PROSPECTING",
org=org_a,
created_by=admin_user,
)
task = Task.objects.create(
title="Patch Clear Opp Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
opportunity=opp,
)
response = admin_client.patch(
f"/api/tasks/{task.id}/",
{"opportunity": ""},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.opportunity_id is None
def test_patch_set_case(self, admin_client, admin_user, org_a):
"""PATCH: set case on a task."""
case = Case.objects.create(
name="Patch Case",
status="New",
priority="High",
org=org_a,
created_by=admin_user,
)
task = Task.objects.create(
title="Patch Case Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.patch(
f"/api/tasks/{task.id}/",
{"case": str(case.id)},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.case_id == case.id
def test_patch_clear_case(self, admin_client, admin_user, org_a):
"""PATCH: clear case."""
case = Case.objects.create(
name="Patch Clear Case",
status="New",
priority="High",
org=org_a,
created_by=admin_user,
)
task = Task.objects.create(
title="Patch Clear Case Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
case=case,
)
response = admin_client.patch(
f"/api/tasks/{task.id}/",
{"case": ""},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.case_id is None
def test_patch_set_lead(self, admin_client, admin_user, org_a):
"""PATCH: set lead on a task."""
lead = Lead.objects.create(
first_name="Patch",
last_name="Lead",
org=org_a,
created_by=admin_user,
)
task = Task.objects.create(
title="Patch Lead Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.patch(
f"/api/tasks/{task.id}/",
{"lead": str(lead.id)},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.lead_id == lead.id
def test_patch_clear_lead(self, admin_client, admin_user, org_a):
"""PATCH: clear lead."""
lead = Lead.objects.create(
first_name="Patch",
last_name="Clear Lead",
org=org_a,
created_by=admin_user,
)
task = Task.objects.create(
title="Patch Clear Lead Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
lead=lead,
)
response = admin_client.patch(
f"/api/tasks/{task.id}/",
{"lead": ""},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.lead_id is None
def test_patch_with_contacts_and_teams(self, admin_client, admin_user, org_a):
"""PATCH with contacts and teams (covers lines 608-637)."""
contact = Contact.objects.create(
first_name="PatchC",
last_name="Contact",
org=org_a,
created_by=admin_user,
)
team = Teams.objects.create(name="Patch Team", org=org_a, created_by=admin_user)
task = Task.objects.create(
title="Patch CT Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.patch(
f"/api/tasks/{task.id}/",
{
"contacts": [str(contact.id)],
"teams": [str(team.id)],
},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.contacts.count() == 1
assert task.teams.count() == 1
def test_patch_with_contacts_as_json_string(self, admin_client, admin_user, org_a):
"""PATCH with contacts as JSON string (covers line 612)."""
contact = Contact.objects.create(
first_name="PatchJSON",
last_name="Contact",
org=org_a,
created_by=admin_user,
)
task = Task.objects.create(
title="Patch JSON Contact Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.patch(
f"/api/tasks/{task.id}/",
{"contacts": json.dumps([str(contact.id)])},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.contacts.count() == 1
def test_patch_with_teams_as_json_string(self, admin_client, admin_user, org_a):
"""PATCH with teams as JSON string (covers line 628)."""
team = Teams.objects.create(
name="Patch JSON Team", org=org_a, created_by=admin_user
)
task = Task.objects.create(
title="Patch JSON Team Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.patch(
f"/api/tasks/{task.id}/",
{"teams": json.dumps([str(team.id)])},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.teams.count() == 1
def test_patch_with_assigned_to_as_json_string(
self, admin_client, admin_profile, org_a
):
"""PATCH with assigned_to as JSON string (covers line 644)."""
task = Task.objects.create(
title="Patch JSON Assigned Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_profile.user,
)
response = admin_client.patch(
f"/api/tasks/{task.id}/",
{"assigned_to": json.dumps([str(admin_profile.id)])},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.assigned_to.count() == 1
def test_patch_with_tags_as_json_string(self, admin_client, admin_user, org_a):
"""PATCH with tags as JSON string (covers line 660)."""
tag = Tags.objects.create(name="patchjsontag", slug="patchjsontag", org=org_a)
task = Task.objects.create(
title="Patch JSON Tag Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
response = admin_client.patch(
f"/api/tasks/{task.id}/",
{"tags": json.dumps([str(tag.id)])},
format="json",
)
assert response.status_code == 200
task.refresh_from_db()
assert task.tags.count() == 1
def test_patch_invalid_returns_400(self, admin_client, admin_user, org_a):
"""PATCH with invalid data returns 400 (covers line 705)."""
task = Task.objects.create(
title="Patch Invalid Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
# Send an invalid status value to trigger serializer error
response = admin_client.patch(
f"/api/tasks/{task.id}/",
{"status": "NotAValidStatus"},
format="json",
)
assert response.status_code == 400
assert response.json()["error"] is True
@pytest.mark.django_db
class TestTaskDetailPostAttachmentCoverage:
"""Tests for POST to task detail with attachment (lines 417-424)."""
def test_post_attachment_to_task(self, admin_client, admin_user, org_a):
"""POST with attachment file to task detail view."""
task = Task.objects.create(
title="Attach Upload Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
upload_file = SimpleUploadedFile(
"testfile.txt", b"file content here", content_type="text/plain"
)
response = admin_client.post(
f"/api/tasks/{task.id}/",
{"task_attachment": upload_file},
format="multipart",
)
assert response.status_code == 200
data = response.json()
assert "attachments" in data
assert len(data["attachments"]) == 1
def test_post_comment_and_attachment(self, admin_client, admin_user, org_a):
"""POST with both comment and attachment.
Note: In multipart mode, the CommentSerializer validation fails because
it requires additional fields (object_id, org, commented_by). So the
comment is NOT saved, but the attachment IS saved. This tests that the
attachment path (lines 417-424) executes.
"""
task = Task.objects.create(
title="Comment And Attach Task",
status="New",
priority="Low",
org=org_a,
created_by=admin_user,
)
upload_file = SimpleUploadedFile(
"doc.pdf", b"pdf content", content_type="application/pdf"
)
response = admin_client.post(
f"/api/tasks/{task.id}/",
{"comment": "With attachment", "task_attachment": upload_file},
format="multipart",
)
assert response.status_code == 200
data = response.json()
# Comment serializer validation fails in multipart mode, so no comment saved
assert len(data["attachments"]) == 1
@pytest.mark.django_db
class TestTaskCommentPatchCoverage:
"""Additional tests for TaskCommentView PATCH and edge cases."""
def _create_task_with_comment(self, admin_user, admin_profile, org_a):
task = Task.objects.create(
title="Comment Coverage Task",
status="New",
priority="Medium",
org=org_a,
created_by=admin_user,
)
ct = ContentType.objects.get_for_model(Task)
comment = Comment.objects.create(
content_type=ct,
object_id=task.id,
comment="Original coverage comment",
commented_by=admin_profile,
org=org_a,
)
return task, comment
def test_patch_comment_non_admin_forbidden(
self, user_client, user_profile, admin_user, admin_profile, org_a
):
"""Non-admin who didn't create comment gets 403 on PATCH (line 827)."""
task, comment = self._create_task_with_comment(admin_user, admin_profile, org_a)
# A reader of the task: the author rule, not the read rule, refuses.
# One who cannot open the task gets the 404 a missing comment gets.
task.assigned_to.add(user_profile)
response = user_client.patch(
f"/api/tasks/comment/{comment.id}/",
{"comment": "Patched by non-admin"},
format="json",
)
assert response.status_code == 403
def test_patch_comment_by_commenter(
self, admin_client, admin_user, admin_profile, org_a
):
"""Commenter (admin profile) can PATCH their own comment."""
_task, comment = self._create_task_with_comment(
admin_user, admin_profile, org_a
)
response = admin_client.patch(
f"/api/tasks/comment/{comment.id}/",
{"comment": "Patched by commenter"},
format="json",
)
assert response.status_code == 200
assert response.json()["message"] == "Comment Updated"