tcms.rpc.utils   A
last analyzed

Complexity

Total Complexity 6

Size/Duplication

Total Lines 98
Duplicated Lines 0 %

Importance

Changes 0
Metric Value
wmc 6
eloc 57
dl 0
loc 98
rs 10
c 0
b 0
f 0

4 Functions

Rating   Name   Duplication   Size   Complexity  
A get_attachments_for() 0 14 2
A add_attachment() 0 11 2
A encode_multipart() 0 27 1
A request_for_upload() 0 25 1
1
# -*- coding: utf-8 -*-
2
3
import io
4
import time
5
6
from attachments import views as attachment_views
7
from attachments.models import Attachment
8
from django.http import HttpRequest
9
from django.middleware.csrf import get_token
10
from mock import MagicMock
11
12
from tcms.core.utils import request_host_link
13
14
15
def get_attachments_for(request, obj):
16
    host_link = request_host_link(request)
17
    result = []
18
    for attachment in Attachment.objects.attachments_for_object(obj):
19
        result.append(
20
            {
21
                "pk": attachment.pk,
22
                "url": host_link + attachment.attachment_file.url,
23
                "owner_pk": attachment.creator.pk,
24
                "owner_username": attachment.creator.username,
25
                "date": attachment.created.isoformat(),
26
            }
27
        )
28
    return result
29
30
31
def encode_multipart(csrf_token, filename, b64content):
32
    """
33
    Build a multipart/form-data body with generated random boundary
34
    suitable for parsing by django.http.request.HttpRequest and
35
    the parser classes related to it!
36
37
    .. note::
38
39
        ``\\r\\n`` are expected! Do not change!
40
    """
41
    boundary = f"----------{int(time.time() * 1000)}"
42
    data = [f"--{boundary}"]
43
44
    data.append('Content-Disposition: form-data; name="csrfmiddlewaretoken"\r\n')
45
    data.append(csrf_token)
46
    data.append(f"--{boundary}")
47
48
    data.append(
49
        f'Content-Disposition: form-data; name="attachment_file"; filename="{filename}"'
50
    )
51
    data.append("Content-Type: application/octet-stream")
52
    data.append("Content-Transfer-Encoding: base64")
53
    data.append(f"Content-Length: {len(b64content)}\r\n")
54
    data.append(b64content)
55
56
    data.append(f"--{boundary}--\r\n")
57
    return "\r\n".join(data), boundary
58
59
60
def request_for_upload(user, filename, b64content):
61
    """
62
    Return a request object containing all fields necessary for file
63
    upload as if it was sent by the browser.
64
    """
65
    request = HttpRequest()
66
    request.user = user
67
    request.method = "POST"
68
    request.content_type = "multipart/form-data"
69
    # because attachment.views.add_attachment() calls messages.success()
70
    request._messages = MagicMock()  # pylint: disable=protected-access
71
72
    data, boundary = encode_multipart(get_token(request), filename, b64content)
73
74
    request.META["CONTENT_TYPE"] = f"multipart/form-data; boundary={boundary}"
75
    request.META["CONTENT_LENGTH"] = len(data)
76
    request._stream = io.BytesIO(data.encode())  # pylint: disable=protected-access
77
78
    # manually parse the input data and populate data attributes
79
    request._read_started = False  # pylint: disable=protected-access
80
    request._load_post_and_files()  # pylint: disable=protected-access
81
    request.POST = request._post  # pylint: disable=protected-access
82
    request.FILES = request._files  # pylint: disable=protected-access
83
84
    return request
85
86
87
def add_attachment(obj_id, app_model, user, filename, b64content):
88
    """
89
    High-level function which performs the attachment process
90
    by constructing an HttpRequest object and passing it to
91
    attachments.views.add_attachment() as if it came from the browser.
92
    """
93
    request = request_for_upload(user, filename, b64content)
94
    app, model = app_model.split(".")
95
    response = attachment_views.add_attachment(request, app, model, obj_id)
96
    if response.status_code == 404:
97
        raise Exception(f"Adding attachment to {app_model}({obj_id}) failed")
98