c3lf-system-3/core/tickets/tests/v2/test_tickets.py

from datetime import datetime, timedelta

from django.test import TestCase, Client

from authentication.models import ExtendedUser
from inventory.models import Event, Container, Item
from mail.models import Email, EmailAttachment
from tickets.models import IssueThread, StateChange, Comment, ItemRelation, Assignment
from django.contrib.auth.models import Permission
from knox.models import AuthToken

from base64 import b64encode


class IssueApiTest(TestCase):

    def setUp(self):
        super().setUp()
        self.user = ExtendedUser.objects.create_user('testuser', 'test', 'test')
        self.user.user_permissions.add(*Permission.objects.all())
        self.user.save()
        self.event = Event.objects.create(slug='evt')
        self.box = Container.objects.create(name='box1')
        self.item = Item.objects.create(container=self.box, description="foo", event=self.event)
        self.token = AuthToken.objects.create(user=self.user)
        self.client = Client(headers={'Authorization': 'Token ' + self.token[1]})

    def test_issues_empty(self):
        response = self.client.get('/api/2/tickets/')
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json(), [])

    def test_issues(self):
        now = datetime.now()
        issue = IssueThread.objects.create(
            name="test issue",
            event=self.event,
        )
        mail1 = Email.objects.create(
            subject='test',
            body='test',
            sender='test',
            recipient='test',
            issue_thread=issue,
            timestamp=now,
        )
        mail2 = Email.objects.create(
            subject='test',
            body='test',
            sender='test',
            recipient='test',
            issue_thread=issue,
            in_reply_to=mail1.reference,
            timestamp=now + timedelta(seconds=2),
        )
        assignment = Assignment.objects.create(
            issue_thread=issue,
            assigned_to=self.user,
            timestamp=now + timedelta(seconds=3),
        )
        comment = Comment.objects.create(
            issue_thread=issue,
            comment="test",
            timestamp=now + timedelta(seconds=4),
        )
        match = ItemRelation.objects.create(
            issue_thread=issue,
            item=self.item,
            timestamp=now + timedelta(seconds=5),
        )
        self.assertEqual('pending_new', issue.state)
        self.assertEqual('test issue', issue.name)
        self.assertEqual(self.user, issue.assigned_to)
        self.assertEqual(36, len(issue.uuid))
        response = self.client.get('/api/2/tickets/')
        self.assertEqual(response.status_code, 200)
        self.assertEqual(len(response.json()), 1)
        self.assertEqual(response.json()[0]['id'], issue.id)
        self.assertEqual(response.json()[0]['name'], "test issue")
        self.assertEqual(response.json()[0]['state'], "pending_new")
        self.assertEqual(response.json()[0]['event'], "evt")
        self.assertEqual(response.json()[0]['assigned_to'], self.user.username)
        self.assertEqual(response.json()[0]['uuid'], issue.uuid)
        self.assertEqual(response.json()[0]['last_activity'], match.timestamp.strftime('%Y-%m-%dT%H:%M:%S.%fZ'))
        self.assertEqual(len(response.json()[0]['timeline']), 6)
        self.assertEqual(response.json()[0]['timeline'][0]['type'], 'state')
        self.assertEqual(response.json()[0]['timeline'][1]['type'], 'mail')
        self.assertEqual(response.json()[0]['timeline'][2]['type'], 'mail')
        self.assertEqual(response.json()[0]['timeline'][3]['type'], 'assignment')
        self.assertEqual(response.json()[0]['timeline'][4]['type'], 'comment')
        self.assertEqual(response.json()[0]['timeline'][5]['type'], 'item_relation')
        self.assertEqual(response.json()[0]['timeline'][1]['id'], mail1.id)
        self.assertEqual(response.json()[0]['timeline'][2]['id'], mail2.id)
        self.assertEqual(response.json()[0]['timeline'][3]['id'], comment.id)
        self.assertEqual(response.json()[0]['timeline'][0]['state'], 'pending_new')
        self.assertEqual(response.json()[0]['timeline'][1]['sender'], 'test')
        self.assertEqual(response.json()[0]['timeline'][1]['recipient'], 'test')
        self.assertEqual(response.json()[0]['timeline'][1]['subject'], 'test')
        self.assertEqual(response.json()[0]['timeline'][1]['body'], 'test')
        self.assertEqual(response.json()[0]['timeline'][1]['timestamp'],
                         mail1.timestamp.strftime('%Y-%m-%dT%H:%M:%S.%fZ'))
        self.assertEqual(response.json()[0]['timeline'][2]['sender'], 'test')
        self.assertEqual(response.json()[0]['timeline'][2]['recipient'], 'test')
        self.assertEqual(response.json()[0]['timeline'][2]['subject'], 'test')
        self.assertEqual(response.json()[0]['timeline'][2]['body'], 'test')
        self.assertEqual(response.json()[0]['timeline'][2]['timestamp'],
                         mail2.timestamp.strftime('%Y-%m-%dT%H:%M:%S.%fZ'))
        self.assertEqual(response.json()[0]['timeline'][3]['assigned_to'], self.user.username)
        self.assertEqual(response.json()[0]['timeline'][3]['timestamp'],
                         assignment.timestamp.strftime('%Y-%m-%dT%H:%M:%S.%fZ'))
        self.assertEqual(response.json()[0]['timeline'][4]['comment'], 'test')
        self.assertEqual(response.json()[0]['timeline'][4]['timestamp'],
                         comment.timestamp.strftime('%Y-%m-%dT%H:%M:%S.%fZ'))
        self.assertEqual(response.json()[0]['timeline'][5]['status'], 'possible')
        self.assertEqual(response.json()[0]['timeline'][5]['timestamp'],
                         match.timestamp.strftime('%Y-%m-%dT%H:%M:%S.%fZ'))
        self.assertEqual(response.json()[0]['timeline'][5]['item']['description'], "foo")
        self.assertEqual(response.json()[0]['timeline'][5]['item']['event'], "evt")
        self.assertEqual(response.json()[0]['timeline'][5]['item']['box'], "box1")
        self.assertEqual(response.json()[0]['related_items'][0]['description'], "foo")
        self.assertEqual(response.json()[0]['related_items'][0]['event'], "evt")
        self.assertEqual(response.json()[0]['related_items'][0]['box'], "box1")

    def test_issues_incomplete_timeline(self):
        now = datetime.now()
        issue1 = IssueThread.objects.create(
            name="test issue",
            event=self.event,
        )
        issue2 = IssueThread.objects.create(
            name="test issue",
            event=self.event,
        )
        issue3 = IssueThread.objects.create(
            name="test issue",
            event=self.event,
        )
        mail1 = Email.objects.create(
            subject='test',
            body='test',
            sender='test',
            recipient='test',
            issue_thread=issue2,
            timestamp=now + timedelta(seconds=2),
        )
        comment = Comment.objects.create(
            issue_thread=issue3,
            comment="test",
            timestamp=now + timedelta(seconds=3),
        )

        response = self.client.get('/api/2/tickets/')
        self.assertEqual(200, response.status_code)
        self.assertEqual(3, len(response.json()))
        self.assertEqual(issue1.id, response.json()[0]['id'])
        self.assertEqual("evt", response.json()[0]['event'])
        self.assertEqual(issue2.id, response.json()[1]['id'])
        self.assertEqual("evt", response.json()[1]['event'])
        self.assertEqual(issue3.id, response.json()[2]['id'])
        self.assertEqual("evt", response.json()[2]['event'])
        self.assertEqual(issue1.state_changes.first().timestamp.strftime('%Y-%m-%dT%H:%M:%S.%fZ'),
                         response.json()[0]['last_activity'])
        self.assertEqual(mail1.timestamp.strftime('%Y-%m-%dT%H:%M:%S.%fZ'),
                         response.json()[1]['last_activity'])
        self.assertEqual(comment.timestamp.strftime('%Y-%m-%dT%H:%M:%S.%fZ'),
                         response.json()[2]['last_activity'])
        self.assertEqual(1, len(response.json()[0]['timeline']))
        self.assertEqual(2, len(response.json()[1]['timeline']))
        self.assertEqual(2, len(response.json()[2]['timeline']))
        self.assertEqual('state', response.json()[0]['timeline'][0]['type'])
        self.assertEqual('state', response.json()[1]['timeline'][0]['type'])
        self.assertEqual('mail', response.json()[1]['timeline'][1]['type'])
        self.assertEqual('state', response.json()[2]['timeline'][0]['type'])
        self.assertEqual('comment', response.json()[2]['timeline'][1]['type'])
        self.assertEqual('pending_new', response.json()[0]['timeline'][0]['state'])
        self.assertEqual('pending_new', response.json()[1]['timeline'][0]['state'])
        self.assertEqual('test', response.json()[1]['timeline'][1]['sender'])
        self.assertEqual('test', response.json()[1]['timeline'][1]['recipient'])
        self.assertEqual('test', response.json()[1]['timeline'][1]['subject'])
        self.assertEqual('test', response.json()[1]['timeline'][1]['body'])
        self.assertEqual(mail1.timestamp.strftime('%Y-%m-%dT%H:%M:%S.%fZ'),
                         response.json()[1]['timeline'][1]['timestamp'])
        self.assertEqual('pending_new', response.json()[2]['timeline'][0]['state'])
        self.assertEqual('test', response.json()[2]['timeline'][1]['comment'])
        self.assertEqual(comment.timestamp.strftime('%Y-%m-%dT%H:%M:%S.%fZ'),
                         response.json()[2]['timeline'][1]['timestamp'])

    def test_issues_with_files(self):
        from django.core.files.base import ContentFile
        from hashlib import sha256
        now = datetime.now()
        issue = IssueThread.objects.create(
            name="test issue",
            event=self.event,
        )
        mail1 = Email.objects.create(
            subject='test',
            body='test',
            sender='test',
            recipient='test',
            issue_thread=issue,
            timestamp=now,
        )
        mail2 = Email.objects.create(
            subject='test',
            body='test',
            sender='test',
            recipient='test',
            issue_thread=issue,
            in_reply_to=mail1.reference,
            timestamp=now + timedelta(seconds=2),
        )
        comment = Comment.objects.create(
            issue_thread=issue,
            comment="test",
            timestamp=now + timedelta(seconds=3),
        )
        file1 = EmailAttachment.objects.create(
            name='file1', mime_type='text/plain', file=ContentFile(b"foo1", "f1"),
            hash=sha256(b"foo1").hexdigest(), email=mail1
        )
        file2 = EmailAttachment.objects.create(
            name='file2', mime_type='text/plain', file=ContentFile(b"foo2", "f2"),
            hash=sha256(b"foo2").hexdigest(), email=mail1
        )

        response = self.client.get('/api/2/tickets/')
        self.assertEqual(200, response.status_code)
        self.assertEqual(1, len(response.json()))
        self.assertEqual(issue.id, response.json()[0]['id'])
        self.assertEqual("evt", response.json()[0]['event'])
        self.assertEqual('pending_new', response.json()[0]['state'])
        self.assertEqual('test issue', response.json()[0]['name'])
        self.assertEqual(None, response.json()[0]['assigned_to'])
        self.assertEqual(36, len(response.json()[0]['uuid']))
        self.assertEqual(comment.timestamp.strftime('%Y-%m-%dT%H:%M:%S.%fZ'),
                         response.json()[0]['last_activity'])
        self.assertEqual(4, len(response.json()[0]['timeline']))
        self.assertEqual('state', response.json()[0]['timeline'][0]['type'])
        self.assertEqual('mail', response.json()[0]['timeline'][1]['type'])
        self.assertEqual('mail', response.json()[0]['timeline'][2]['type'])
        self.assertEqual('comment', response.json()[0]['timeline'][3]['type'])
        self.assertEqual(mail1.id, response.json()[0]['timeline'][1]['id'])
        self.assertEqual(mail2.id, response.json()[0]['timeline'][2]['id'])
        self.assertEqual(comment.id, response.json()[0]['timeline'][3]['id'])
        self.assertEqual('pending_new', response.json()[0]['timeline'][0]['state'])
        self.assertEqual('test', response.json()[0]['timeline'][1]['sender'])
        self.assertEqual('test', response.json()[0]['timeline'][1]['recipient'])
        self.assertEqual('test', response.json()[0]['timeline'][1]['subject'])
        self.assertEqual('test', response.json()[0]['timeline'][1]['body'])
        self.assertEqual(mail1.timestamp.strftime('%Y-%m-%dT%H:%M:%S.%fZ'),
                         response.json()[0]['timeline'][1]['timestamp'])
        self.assertEqual('test', response.json()[0]['timeline'][2]['sender'])
        self.assertEqual('test', response.json()[0]['timeline'][2]['recipient'])

        self.assertEqual('test', response.json()[0]['timeline'][2]['subject'])
        self.assertEqual('test', response.json()[0]['timeline'][2]['body'])
        self.assertEqual(mail2.timestamp.strftime('%Y-%m-%dT%H:%M:%S.%fZ'),
                         response.json()[0]['timeline'][2]['timestamp'])
        self.assertEqual('test', response.json()[0]['timeline'][3]['comment'])
        self.assertEqual(comment.timestamp.strftime('%Y-%m-%dT%H:%M:%S.%fZ'),
                         response.json()[0]['timeline'][3]['timestamp'])
        self.assertEqual(2, len(response.json()[0]['timeline'][1]['attachments']))
        self.assertEqual(0, len(response.json()[0]['timeline'][2]['attachments']))
        self.assertEqual('file1', response.json()[0]['timeline'][1]['attachments'][0]['name'])
        self.assertEqual('file2', response.json()[0]['timeline'][1]['attachments'][1]['name'])
        self.assertEqual('text/plain', response.json()[0]['timeline'][1]['attachments'][0]['mime_type'])
        self.assertEqual('text/plain', response.json()[0]['timeline'][1]['attachments'][1]['mime_type'])
        self.assertEqual(file1.hash, response.json()[0]['timeline'][1]['attachments'][0]['hash'])
        self.assertEqual(file2.hash, response.json()[0]['timeline'][1]['attachments'][1]['hash'])

    def test_manual_creation(self):
        response = self.client.post('/api/2/evt/tickets/manual/',
                                    {'name': 'test issue', 'sender': 'test', 'recipient': 'test', 'body': 'test'},
                                    content_type='application/json')
        self.assertEqual(response.status_code, 201)
        self.assertEqual(response.json()['state'], 'pending_new')
        self.assertEqual(response.json()['name'], 'test issue')
        self.assertEqual(response.json()['assigned_to'], None)
        self.assertEqual("evt", response.json()['event'])
        timeline = response.json()['timeline']
        self.assertEqual(len(timeline), 2)
        self.assertEqual(timeline[0]['type'], 'state')
        self.assertEqual(timeline[0]['state'], 'pending_new')
        self.assertEqual(timeline[1]['type'], 'mail')
        self.assertEqual(timeline[1]['sender'], 'test')
        self.assertEqual(timeline[1]['recipient'], 'test')
        self.assertEqual(timeline[1]['subject'], 'test issue')
        self.assertEqual(timeline[1]['body'], 'test')

    def test_manual_creation_none(self):
        response = self.client.post('/api/2/none/tickets/manual/',
                                    {'name': 'test issue', 'sender': 'test', 'recipient': 'test', 'body': 'test'},
                                    content_type='application/json')
        self.assertEqual(response.status_code, 201)
        self.assertEqual(response.json()['state'], 'pending_new')
        self.assertEqual(response.json()['name'], 'test issue')
        self.assertEqual(response.json()['assigned_to'], None)
        self.assertEqual(None, response.json()['event'])
        timeline = response.json()['timeline']
        self.assertEqual(len(timeline), 2)
        self.assertEqual(timeline[0]['type'], 'state')
        self.assertEqual(timeline[0]['state'], 'pending_new')
        self.assertEqual(timeline[1]['type'], 'mail')
        self.assertEqual(timeline[1]['sender'], 'test')
        self.assertEqual(timeline[1]['recipient'], 'test')
        self.assertEqual(timeline[1]['subject'], 'test issue')
        self.assertEqual(timeline[1]['body'], 'test')

    def test_manual_creation_invalid(self):
        response = self.client.post('/api/2/foobar/tickets/manual/',
                                    {'name': 'test issue', 'sender': 'test', 'recipient': 'test', 'body': 'test'},
                                    content_type='application/json')
        self.assertEqual(response.status_code, 400)

    #def test_post_comment(self):
    #    issue = IssueThread.objects.create(
    #        name="test issue",
    #    )
    #    response = self.client.post('/api/2/comments/', {'comment': 'test', 'issue_thread': issue.id})
    #    self.assertEqual(response.status_code, 201)
    #    self.assertEqual(response.json()['comment'], 'test')
    #    self.assertEqual(response.json()['issue_thread'], issue.id)
    #    self.assertEqual(response.json()['timestamp'], response.json()['timestamp'])

    def test_post_comment_altenative(self):
        issue = IssueThread.objects.create(
            name="test issue",
            event=self.event,
        )
        response = self.client.post(f'/api/2/tickets/{issue.id}/comment/', {'comment': 'test'})
        self.assertEqual(response.status_code, 201)
        self.assertEqual(response.json()['comment'], 'test')
        self.assertEqual(response.json()['issue_thread'], issue.id)
        self.assertEqual(response.json()['timestamp'], response.json()['timestamp'])

    def test_post_alt_comment_empty(self):
        issue = IssueThread.objects.create(
            name="test issue",
            event=self.event,
        )
        response = self.client.post(f'/api/2/tickets/{issue.id}/comment/', {'comment': ''})
        self.assertEqual(response.status_code, 400)

    def test_state_change(self):
        issue = IssueThread.objects.create(
            name="test issue",
            event=self.event,
        )
        response = self.client.patch(f'/api/2/tickets/{issue.id}/', {'state': 'pending_open'},
                                     content_type='application/json')
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()['state'], 'pending_open')
        self.assertEqual(response.json()['name'], 'test issue')
        self.assertEqual(response.json()['assigned_to'], None)
        timeline = response.json()['timeline']
        self.assertEqual(len(timeline), 2)
        self.assertEqual(timeline[0]['type'], 'state')
        self.assertEqual(timeline[0]['state'], 'pending_new')
        self.assertEqual(timeline[1]['type'], 'state')
        self.assertEqual(timeline[1]['state'], 'pending_open')

    def test_state_change_invalid_state(self):
        issue = IssueThread.objects.create(
            name="test issue",
            event=self.event,
        )
        response = self.client.patch(f'/api/2/tickets/{issue.id}/', {'state': 'invalid'},
                                     content_type='application/json')
        self.assertEqual(400, response.status_code)

    def test_assign_user(self):
        issue = IssueThread.objects.create(
            name="test issue",
            event=self.event,
        )
        response = self.client.patch(f'/api/2/tickets/{issue.id}/', {'assigned_to': self.user.username},
                                     content_type='application/json')
        self.assertEqual(200, response.status_code)
        self.assertEqual('pending_new', response.json()['state'])
        self.assertEqual('test issue', response.json()['name'])
        self.assertEqual("evt", response.json()['event'])
        self.assertEqual(self.user.username, response.json()['assigned_to'])
        timeline = response.json()['timeline']
        self.assertEqual(2, len(timeline))
        self.assertEqual('state', timeline[0]['type'])
        self.assertEqual('pending_new', timeline[0]['state'])
        self.assertEqual('assignment', timeline[1]['type'])
        self.assertEqual(self.user.username, timeline[1]['assigned_to'])


class IssueSearchTest(TestCase):

    def setUp(self):
        super().setUp()
        self.event = Event.objects.create(slug='EVENT', name='Event')
        self.user = ExtendedUser.objects.create_user('testuser', 'test', 'test')
        self.user.user_permissions.add(*Permission.objects.all())
        self.user.save()
        self.token = AuthToken.objects.create(user=self.user)
        self.client = Client(headers={'Authorization': 'Token ' + self.token[1]})

    def test_search(self):
        search_query = b64encode(b'abc').decode('utf-8')
        response = self.client.get(f'/api/2/{self.event.slug}/tickets/{search_query}/')
        self.assertEqual(200, response.status_code)
        self.assertEqual([], response.json())