update tests for Flask
This commit is contained in:
@@ -1,24 +1,25 @@
|
||||
from sqlalchemy.dialects.postgresql import insert
|
||||
|
||||
from atst.database import db
|
||||
from atst.models.pe_number import PENumber
|
||||
from .exceptions import NotFoundError
|
||||
|
||||
|
||||
class PENumbers(object):
|
||||
def __init__(self, db_session):
|
||||
self.db_session = db_session
|
||||
|
||||
def get(self, number):
|
||||
pe_number = self.db_session.query(PENumber).get(number)
|
||||
@classmethod
|
||||
def get(cls, number):
|
||||
pe_number = db.session.query(PENumber).get(number)
|
||||
if not pe_number:
|
||||
raise NotFoundError("pe_number")
|
||||
|
||||
return pe_number
|
||||
|
||||
def create_many(self, list_of_pe_numbers):
|
||||
@classmethod
|
||||
def create_many(cls, list_of_pe_numbers):
|
||||
stmt = insert(PENumber).values(list_of_pe_numbers)
|
||||
do_update = stmt.on_conflict_do_update(
|
||||
index_elements=["number"], set_=dict(description=stmt.excluded.description)
|
||||
)
|
||||
self.db_session.execute(do_update)
|
||||
self.db_session.commit()
|
||||
db.session.execute(do_update)
|
||||
db.session.commit()
|
||||
|
@@ -1,12 +1,11 @@
|
||||
import re
|
||||
import tornado
|
||||
from tornado.gen import Return
|
||||
from wtforms.fields.html5 import EmailField
|
||||
from wtforms.fields import StringField, SelectField
|
||||
from wtforms.form import Form
|
||||
from wtforms.validators import Required, Email
|
||||
|
||||
from atst.domain.exceptions import NotFoundError
|
||||
from atst.domain.pe_numbers import PENumbers
|
||||
|
||||
from .fields import NewlineListField
|
||||
from .forms import ValidatedForm
|
||||
@@ -40,9 +39,9 @@ def suggest_pe_id(pe_id):
|
||||
return None
|
||||
|
||||
|
||||
def validate_pe_id(field, existing_request, pe_numbers_repo):
|
||||
def validate_pe_id(field, existing_request):
|
||||
try:
|
||||
pe_number = pe_numbers_repo.get(field.data)
|
||||
pe_number = PENumbers.get(field.data)
|
||||
except NotFoundError:
|
||||
suggestion = suggest_pe_id(field.data)
|
||||
error_str = (
|
||||
@@ -50,17 +49,17 @@ def validate_pe_id(field, existing_request, pe_numbers_repo):
|
||||
"If you have double checked it you can submit anyway. "
|
||||
"Your request will need to go through a manual review."
|
||||
).format('Did you mean "{}"? '.format(suggestion) if suggestion else "")
|
||||
field.errors.append(error_str)
|
||||
field.errors += (error_str,)
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
class FinancialForm(ValidatedForm):
|
||||
def perform_extra_validation(self, existing_request, pe_numbers_repo):
|
||||
def perform_extra_validation(self, existing_request):
|
||||
valid = True
|
||||
if not existing_request or existing_request.get("pe_id") != self.pe_id.data:
|
||||
valid = yield validate_pe_id(self.pe_id, existing_request, pe_numbers_repo)
|
||||
valid = validate_pe_id(self.pe_id, existing_request)
|
||||
return valid
|
||||
|
||||
task_order_id = StringField(
|
||||
|
@@ -7,6 +7,11 @@ bp = Blueprint("atst", __name__)
|
||||
|
||||
|
||||
@bp.route("/")
|
||||
def root():
|
||||
return render_template("root.html")
|
||||
|
||||
|
||||
@bp.route("/home")
|
||||
def home():
|
||||
return render_template("home.html")
|
||||
|
||||
@@ -14,3 +19,8 @@ def home():
|
||||
@bp.route("/styleguide")
|
||||
def styleguide():
|
||||
return render_template("styleguide.html")
|
||||
|
||||
|
||||
@bp.route('/<path:path>')
|
||||
def catch_all(path):
|
||||
return render_template("{}.html".format(path))
|
||||
|
@@ -1,52 +0,0 @@
|
||||
from flask import Blueprint, g, render_template
|
||||
import pendulum
|
||||
|
||||
from atst.domain.requests import Requests
|
||||
|
||||
requests_bp = Blueprint("requests", __name__)
|
||||
|
||||
def map_request(user, request):
|
||||
time_created = pendulum.instance(request.time_created)
|
||||
is_new = time_created.add(days=1) > pendulum.now()
|
||||
|
||||
return {
|
||||
"order_id": request.id,
|
||||
"is_new": is_new,
|
||||
"status": request.status,
|
||||
"app_count": 1,
|
||||
"date": time_created.format("M/DD/YYYY"),
|
||||
"full_name": "{} {}".format(user["first_name"], user["last_name"]),
|
||||
}
|
||||
|
||||
|
||||
@requests_bp.route("/requests", methods=["GET"])
|
||||
def requests_index():
|
||||
requests = []
|
||||
if "review_and_approve_jedi_workspace_request" in g.current_user["atat_permissions"]:
|
||||
requests = Requests.get_many()
|
||||
else:
|
||||
requests = Requests.get_many(creator_id=g.current_user["id"])
|
||||
|
||||
mapped_requests = [map_request(g.current_user, r) for r in requests]
|
||||
|
||||
return render_template("requests.html", requests=mapped_requests)
|
||||
|
||||
|
||||
@requests_bp.route("/requests/new/<int:screen>", methods=["GET"])
|
||||
def requests_new():
|
||||
pass
|
||||
|
||||
|
||||
@requests_bp.route("/requests/new/<int:screen>/<string:request_id>", methods=["GET"])
|
||||
def requests_form_update():
|
||||
pass
|
||||
|
||||
|
||||
@requests_bp.route("/requests/verify/<string:request_id>", methods=["GET"])
|
||||
def financial_verification():
|
||||
pass
|
||||
|
||||
|
||||
@requests_bp.route("/requests/verify/<string:request_id>", methods=["POST"])
|
||||
def update_financial_verification():
|
||||
pass
|
@@ -1,4 +1,5 @@
|
||||
from flask import render_template, redirect, url_for
|
||||
from flask import request as http_request
|
||||
|
||||
from . import requests_bp
|
||||
from atst.domain.requests import Requests
|
||||
@@ -29,7 +30,7 @@ def update_financial_verification(request_id):
|
||||
existing_request.body.get("financial_verification")
|
||||
)
|
||||
if valid:
|
||||
redirect(url_for("requests.financial_verification_submitted"))
|
||||
return redirect(url_for("requests.financial_verification_submitted"))
|
||||
else:
|
||||
return render_template(
|
||||
"requests/financial_verification.html", **rerender_args
|
||||
|
Reference in New Issue
Block a user