testrail_bot/control/views.py

import datetime
import json
import os
from .celery_tasks import test_rail_api

from django.shortcuts import render, redirect, HttpResponse

from . import models
from . import forms
from .celery_tasks.tasks import process_run, update_plot_data


def index(request):
    runs = models.TestRailTestRun.objects.all()
    return render(request, "control/index.html", {"runs": runs})


def redirect_to_index(request):
    return redirect("index")


def single_run(request, run_id):
    run = models.TestRailTestRun.objects.get(pk=run_id)
    if request.method == "POST":
        form = forms.TestRunForm(request.POST, instance=run)
        if form.is_valid():
            form.save()
            return redirect("single_run", run_id)
    else:
        form = forms.TestRunForm(instance=run)

    return render(request, "control/update_run.html",
                  {"form": form, "run_id": run_id})


def create_run(request):
    if request.method == "POST":
        form = forms.TestRunForm(request.POST)
        if form.is_valid():
            obj = form.save()
            return redirect("single_run", obj.id)
    else:
        form = forms.TestRunForm()

    return render(request, "control/create_run.html", {"form": form})


def list_reports(request):
    reports = models.TestRailReport.objects.order_by("-created_at").all()
    return render(request, "control/reports.html", {"reports": reports})


def single_report(request, report_id):
    report = models.TestRailReport.objects.get(pk=report_id)
    data = report.path.read().decode("utf-8")
    if request.method == "POST" and request.is_ajax():
        return HttpResponse(
            json.dumps({"data": data, "finished": report.finished}),
            content_type="application/json")
    return render(request, "control/report.html",
                  {"report_id": report.id, "report": data})


def submit_run(request, run_id):
    run = models.TestRailTestRun.objects.get(pk=run_id)
    testrail_run = test_rail_api.get_run_by_id(run.run_id)
    if not run.run_name:
        run.run_name = testrail_run['name']
    report_name = "{}-run_id-{}-date-{}".format(
        run.run_name, run.run_id, datetime.datetime.isoformat(datetime.datetime.now()))
    path = os.path.join(models.fs.location, report_name)
    with open(path, "w"):
        pass
    form = forms.TestRunForm(request.POST)
    run_date = None
    if form.is_valid():
        run_date = datetime.datetime.combine(
            form.cleaned_data["timestamp"] + datetime.timedelta(days=1),
            datetime.datetime.min.time()).timestamp()

    report = models.TestRailReport(
        test_run=run,
        report_name=report_name,
        path=path)
    report.save()
    process_run.delay(run_id, report.id, path, run_date)

    return redirect("single_report", report.id)


def show_help(request):
    return render(request, "control/help.html")


def update_jenkins_plot(request):
    try:
        models.ActionLog.objects.get(name="update_plot_started")
        return HttpResponse("Update in progress", status=403)
    except models.ActionLog.DoesNotExist:
        pass
    update = models.ActionLog(
        name="update_plot_started", date=datetime.datetime.now())
    update.save()
    update_plot_data.delay()
    return HttpResponse("Started Update", status=200)


def jenkins_plot(request):
    try:
        update_date = models.ActionLog.objects.get(
            name="update_jenkins_plot").date
    except models.ActionLog.DoesNotExist:
        update_date = None
    try:
        models.ActionLog.objects.get(name="update_plot_started")
        update_started = True
    except models.ActionLog.DoesNotExist:
        update_started = False

    job_names_path = os.path.join(models.fs.location, "job_names.txt")
    job_names = []
    if os.path.exists(job_names_path):
        try:
            with open(job_names_path, "r") as f:
                job_names = json.load(f)
        except:
            pass

    return render(
        request, "control/jenkins_plot.html",
        {"update_date": update_date, "update_started": update_started,
         "job_names": enumerate(job_names, 1)})