<commit_before>#  engagements
import logging
from datetime import datetime
import operator
import base64
from django.contrib.auth.models import User
from django.conf import settings
from django.contrib import messages
from django.contrib.auth.decorators import user_passes_test
from django.core.exceptions import PermissionDenied
from django.urls import reverse
from django.db.models import Q, Count
from django.http import HttpResponseRedirect, StreamingHttpResponse, Http404, HttpResponse, FileResponse
from django.shortcuts import render, get_object_or_404
from django.views.decorators.cache import cache_page
from django.utils import timezone
from time import strftime
from django.contrib.admin.utils import NestedObjects
from django.db import DEFAULT_DB_ALIAS
from django.core.exceptions import MultipleObjectsReturned

from dojo.engagement.services import close_engagement, reopen_engagement
from dojo.filters import EngagementFilter, EngagementTestFilter
from dojo.forms import CheckForm, \
    UploadThreatForm, RiskAcceptanceForm, NoteForm, DoneForm, \
    EngForm, TestForm, ReplaceRiskAcceptanceProofForm, AddFindingsRiskAcceptanceForm, DeleteEngagementForm, ImportScanForm, \
    CredMappingForm, JIRAEngagementForm, JIRAImportScanForm, TypedNoteForm, JIRAProjectForm, \
    EditRiskAcceptanceForm

from dojo.models import Finding, IMPORT_CREATED_FINDING, Product, Engagement, Test, \
    Check_List, Test_Import, Test_Import_Finding_Action, Test_Type, Notes, \
    Risk_Acceptance, Development_Environment, BurpRawRequestResponse, Endpoint, \
    Cred_Mapping, Dojo_User, System_Settings, Note_Type, Endpoint_Status
from dojo.tools.factory import handles_active_verified_statuses
from dojo.tools.factory import import_parser_factory, get_choices
from dojo.utils import get_page_items, add_breadcrumb, handle_uploaded_threat, \
    FileIterWrapper, get_cal_event, message, Product_Tab, is_scan_file_too_large, \
    get_system_setting, redirect_to_return_url_or_else, get_return_url
from dojo.notifications.helper import create_notification
from dojo.finding.views import find_available_notetypes
from functools import reduce
from django.db.models.query import QuerySet
from dojo.user.helper import user_must_be_authorized, user_is_authorized, check_auth_users_list
import dojo.jira_link.helper as jira_helper
import dojo.risk_acceptance.helper as ra_helper
from dojo.risk_acceptance.helper import prefetch_for_expiration
from dojo.finding.views import NOT_ACCEPTED_FINDINGS_QUERY
from django.views.decorators.vary import vary_on_cookie


logger = logging.getLogger(__name__)
parse_logger = logging.getLogger('dojo')


@user_passes_test(lambda u: u.is_staff)
@cache_page(60 * 5)  # cache for 5 minutes
@vary_on_cookie
def engagement_calendar(request):
    if 'lead' not in request.GET or '0' in request.GET.getlist('lead'):
        engagements = Engagement.objects.all()
    else:
        filters = []
        leads = request.GET.getlist('lead', '')
        if '-1' in request.GET.getlist('lead'):
            leads.remove('-1')
            filters.append(Q(lead__isnull=True))
        filters.append(Q(lead__in=leads))
        engagements = Engagement.objects.filter(reduce(operator.or_, filters))

    engagements = engagements.select_related('lead')
    engagements = engagements.prefetch_related('product')

    add_breadcrumb(
        title="Engagement Calendar", top_level=True, request=request)
    return render(
        request, 'dojo/calendar.html', {
            'caltype': 'engagements',
            'leads': request.GET.getlist('lead', ''),
            'engagements': engagements,
            'users': Dojo_User.objects.all()
        })


@user_passes_test(lambda u: u.is_staff)
def engagement(request):
    products_with_engagements = Product.objects.filter(~Q(engagement=None), engagement__active=True).distinct()
    filtered = EngagementFilter(
        request.GET,
        queryset=products_with_engagements.prefetch_related('engagement_set', 'prod_type', 'engagement_set__lead',
                                                            'engagement_set__test_set__lead', 'engagement_set__test_set__test_type'))
    prods = get_page_items(request, filtered.qs, 25)
    name_words = products_with_engagements.values_list('name', flat=True)
    eng_words = Engagement.objects.filter(active=True).values_list('name', flat=True).distinct()

    add_breadcrumb(
        title="Active Engagements",
        top_level=not len(request.GET),
        request=request)

    prods.object_list = prefetch_for_products_with_engagments(prods.object_list)

    return render(
        request, 'dojo/engagement.html', {
            'products': prods,
            'filtered': filtered,
            'name_words': sorted(set(name_words)),
            'eng_words': sorted(set(eng_words)),
        })


@user_passes_test(lambda u: u.is_staff)
def engagements_all(request):

    products_with_engagements = Product.objects.filter(~Q(engagement=None)).distinct()
    filtered = EngagementFilter(
        request.GET,
        queryset=products_with_engagements.prefetch_related('engagement_set', 'prod_type', 'engagement_set__lead',
                                                            'engagement_set__test_set__lead', 'engagement_set__test_set__test_type'))

    prods = get_page_items(request, filtered.qs, 25)

    name_words = products_with_engagements.values_list('name', flat=True)
    eng_words = Engagement.objects.all().values_list('name', flat=True).distinct()

    add_breadcrumb(
        title="All Engagements",
        top_level=not len(request.GET),
        request=request)

    prods.object_list = prefetch_for_products_with_engagments(prods.object_list)

    return render(
        request, 'dojo/engagements_all.html', {
            'products': prods,
            'filtered': filtered,
            'name_words': sorted(set(name_words)),
            'eng_words': sorted(set(eng_words)),
        })


def prefetch_for_products_with_engagments(products_with_engagements):
    if isinstance(products_with_engagements, QuerySet):  # old code can arrive here with prods being a list because the query was already executed
        return products_with_engagements.prefetch_related('tags',
            'engagement_set__tags',
            'engagement_set__test_set__tags',
            'engagement_set__jira_project__jira_instance',
            'jira_project_set__jira_instance')

    logger.debug('unable to prefetch because query was already executed')
    return products_with_engagements


@user_must_be_authorized(Engagement, 'change', 'eid')
def edit_engagement(request, eid):
    engagement = Engagement.objects.get(pk=eid)
    is_ci_cd = engagement.engagement_type == "CI/CD"
    jira_epic_form = None
    jira_project = None
    jira_error = False

    if request.method == 'POST':
        form = EngForm(request.POST, instance=engagement, cicd=is_ci_cd, product=engagement.product.id, user=request.user)
        jira_project = jira_helper.get_jira_project(engagement, use_inheritance=False)

        if form.is_valid():
            # first save engagement details
            new_status = form.cleaned_data.get('status')
            engagement = form.save(commit=False)
            if (new_status == "Cancelled" or new_status == "Completed"):
                engagement.active = False
                create_notification(event='close_engagement',
                        title='Closure of %s' % engagement.name,
                        description='The engagement "%s" was closed' % (engagement.name),
                        engagement=engagaement, url=reverse('engagment_all_findings', args=(engagement.id, ))),
            else:
                engagement.active = True
            engagement.save()
            form.save_m2m()

            messages.add_message(
                request,
                messages.SUCCESS,
                'Engagement updated successfully.',
                extra_tags='alert-success')

            success, jira_project_form = jira_helper.process_jira_project_form(request, instance=jira_project, engagement=engagement)
            error = not success

            success, jira_epic_form = jira_helper.process_jira_epic_form(request, engagement=engagement)
            error = error or not success

            if not error:
                if '_Add Tests' in request.POST:
                    return HttpResponseRedirect(
                        reverse('add_tests', args=(engagement.id, )))
                else:
                    return HttpResponseRedirect(
                        reverse('view_engagement', args=(engagement.id, )))
        else:
            logger.debug(form.errors)

    form = EngForm(initial={'product': engagement.product}, instance=engagement, cicd=is_ci_cd, product=engagement.product, user=request.user)

    jira_project_form = None
    jira_epic_form = None<commit_msg>fix typo (#3895)<commit_after>#  engagements
import logging
from datetime import datetime
import operator
import base64
from django.contrib.auth.models import User
from django.conf import settings
from django.contrib import messages
from django.contrib.auth.decorators import user_passes_test
from django.core.exceptions import PermissionDenied
from django.urls import reverse
from django.db.models import Q, Count
from django.http import HttpResponseRedirect, StreamingHttpResponse, Http404, HttpResponse, FileResponse
from django.shortcuts import render, get_object_or_404
from django.views.decorators.cache import cache_page
from django.utils import timezone
from time import strftime
from django.contrib.admin.utils import NestedObjects
from django.db import DEFAULT_DB_ALIAS
from django.core.exceptions import MultipleObjectsReturned

from dojo.engagement.services import close_engagement, reopen_engagement
from dojo.filters import EngagementFilter, EngagementTestFilter
from dojo.forms import CheckForm, \
    UploadThreatForm, RiskAcceptanceForm, NoteForm, DoneForm, \
    EngForm, TestForm, ReplaceRiskAcceptanceProofForm, AddFindingsRiskAcceptanceForm, DeleteEngagementForm, ImportScanForm, \
    CredMappingForm, JIRAEngagementForm, JIRAImportScanForm, TypedNoteForm, JIRAProjectForm, \
    EditRiskAcceptanceForm

from dojo.models import Finding, IMPORT_CREATED_FINDING, Product, Engagement, Test, \
    Check_List, Test_Import, Test_Import_Finding_Action, Test_Type, Notes, \
    Risk_Acceptance, Development_Environment, BurpRawRequestResponse, Endpoint, \
    Cred_Mapping, Dojo_User, System_Settings, Note_Type, Endpoint_Status
from dojo.tools.factory import handles_active_verified_statuses
from dojo.tools.factory import import_parser_factory, get_choices
from dojo.utils import get_page_items, add_breadcrumb, handle_uploaded_threat, \
    FileIterWrapper, get_cal_event, message, Product_Tab, is_scan_file_too_large, \
    get_system_setting, redirect_to_return_url_or_else, get_return_url
from dojo.notifications.helper import create_notification
from dojo.finding.views import find_available_notetypes
from functools import reduce
from django.db.models.query import QuerySet
from dojo.user.helper import user_must_be_authorized, user_is_authorized, check_auth_users_list
import dojo.jira_link.helper as jira_helper
import dojo.risk_acceptance.helper as ra_helper
from dojo.risk_acceptance.helper import prefetch_for_expiration
from dojo.finding.views import NOT_ACCEPTED_FINDINGS_QUERY
from django.views.decorators.vary import vary_on_cookie


logger = logging.getLogger(__name__)
parse_logger = logging.getLogger('dojo')


@user_passes_test(lambda u: u.is_staff)
@cache_page(60 * 5)  # cache for 5 minutes
@vary_on_cookie
def engagement_calendar(request):
    if 'lead' not in request.GET or '0' in request.GET.getlist('lead'):
        engagements = Engagement.objects.all()
    else:
        filters = []
        leads = request.GET.getlist('lead', '')
        if '-1' in request.GET.getlist('lead'):
            leads.remove('-1')
            filters.append(Q(lead__isnull=True))
        filters.append(Q(lead__in=leads))
        engagements = Engagement.objects.filter(reduce(operator.or_, filters))

    engagements = engagements.select_related('lead')
    engagements = engagements.prefetch_related('product')

    add_breadcrumb(
        title="Engagement Calendar", top_level=True, request=request)
    return render(
        request, 'dojo/calendar.html', {
            'caltype': 'engagements',
            'leads': request.GET.getlist('lead', ''),
            'engagements': engagements,
            'users': Dojo_User.objects.all()
        })


@user_passes_test(lambda u: u.is_staff)
def engagement(request):
    products_with_engagements = Product.objects.filter(~Q(engagement=None), engagement__active=True).distinct()
    filtered = EngagementFilter(
        request.GET,
        queryset=products_with_engagements.prefetch_related('engagement_set', 'prod_type', 'engagement_set__lead',
                                                            'engagement_set__test_set__lead', 'engagement_set__test_set__test_type'))
    prods = get_page_items(request, filtered.qs, 25)
    name_words = products_with_engagements.values_list('name', flat=True)
    eng_words = Engagement.objects.filter(active=True).values_list('name', flat=True).distinct()

    add_breadcrumb(
        title="Active Engagements",
        top_level=not len(request.GET),
        request=request)

    prods.object_list = prefetch_for_products_with_engagments(prods.object_list)

    return render(
        request, 'dojo/engagement.html', {
            'products': prods,
            'filtered': filtered,
            'name_words': sorted(set(name_words)),
            'eng_words': sorted(set(eng_words)),
        })


@user_passes_test(lambda u: u.is_staff)
def engagements_all(request):

    products_with_engagements = Product.objects.filter(~Q(engagement=None)).distinct()
    filtered = EngagementFilter(
        request.GET,
        queryset=products_with_engagements.prefetch_related('engagement_set', 'prod_type', 'engagement_set__lead',
                                                            'engagement_set__test_set__lead', 'engagement_set__test_set__test_type'))

    prods = get_page_items(request, filtered.qs, 25)

    name_words = products_with_engagements.values_list('name', flat=True)
    eng_words = Engagement.objects.all().values_list('name', flat=True).distinct()

    add_breadcrumb(
        title="All Engagements",
        top_level=not len(request.GET),
        request=request)

    prods.object_list = prefetch_for_products_with_engagments(prods.object_list)

    return render(
        request, 'dojo/engagements_all.html', {
            'products': prods,
            'filtered': filtered,
            'name_words': sorted(set(name_words)),
            'eng_words': sorted(set(eng_words)),
        })


def prefetch_for_products_with_engagments(products_with_engagements):
    if isinstance(products_with_engagements, QuerySet):  # old code can arrive here with prods being a list because the query was already executed
        return products_with_engagements.prefetch_related('tags',
            'engagement_set__tags',
            'engagement_set__test_set__tags',
            'engagement_set__jira_project__jira_instance',
            'jira_project_set__jira_instance')

    logger.debug('unable to prefetch because query was already executed')
    return products_with_engagements


@user_must_be_authorized(Engagement, 'change', 'eid')
def edit_engagement(request, eid):
    engagement = Engagement.objects.get(pk=eid)
    is_ci_cd = engagement.engagement_type == "CI/CD"
    jira_epic_form = None
    jira_project = None
    jira_error = False

    if request.method == 'POST':
        form = EngForm(request.POST, instance=engagement, cicd=is_ci_cd, product=engagement.product.id, user=request.user)
        jira_project = jira_helper.get_jira_project(engagement, use_inheritance=False)

        if form.is_valid():
            # first save engagement details
            new_status = form.cleaned_data.get('status')
            engagement = form.save(commit=False)
            if (new_status == "Cancelled" or new_status == "Completed"):
                engagement.active = False
                create_notification(event='close_engagement',
                        title='Closure of %s' % engagement.name,
                        description='The engagement "%s" was closed' % (engagement.name),
                        engagement=engagement, url=reverse('engagment_all_findings', args=(engagement.id, ))),
            else:
                engagement.active = True
            engagement.save()
            form.save_m2m()

            messages.add_message(
                request,
                messages.SUCCESS,
                'Engagement updated successfully.',
                extra_tags='alert-success')

            success, jira_project_form = jira_helper.process_jira_project_form(request, instance=jira_project, engagement=engagement)
            error = not success

            success, jira_epic_form = jira_helper.process_jira_epic_form(request, engagement=engagement)
            error = error or not success

            if not error:
                if '_Add Tests' in request.POST:
                    return HttpResponseRedirect(
                        reverse('add_tests', args=(engagement.id, )))
                else:
                    return HttpResponseRedirect(
                        reverse('view_engagement', args=(engagement.id, )))
        else:
            logger.debug(form.errors)

    form = EngForm(initial={'product': engagement.product}, instance=engagement, cicd=is_ci_cd, product=engagement.product, user=request.user)

    jira_project_form = None
    jira_epic_form = None
{'fraction_non_alphanumeric': 0.08771830985915494, 'fraction_numerical': 0.0024788732394366198, 'mean_word_length': 3.4859742228961332, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5490493', 'n_tokens_mistral': 5255, 'n_tokens_neox': 5084, 'n_words': 1011}