pk/atst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

remove tornado handlers

Browse Source
This commit is contained in:
dandds 2018-08-06 13:16:08 -04:00 committed by richard-dds
parent 15f59971bd
commit 8587ad4e4b
13 changed files with 0 additions and 574 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
atst/handler.py
View File

@ -1,42 +0,0 @@
import tornado.web
from atst.sessions import SessionNotFoundError
from atst.domain.users import Users
helpers = {"assets": None}
class BaseHandler(tornado.web.RequestHandler):
def get_template_namespace(self):
ns = super(BaseHandler, self).get_template_namespace()
helpers["config"] = self.application.config
ns.update(helpers)
return ns
@tornado.gen.coroutine
def login(self, user):
db_user = yield self._get_user_permissions(user["id"])
user["atat_permissions"] = db_user.atat_permissions
user["atat_role"] = db_user.atat_role.name
session_id = self.sessions.start_session(user)
self.set_secure_cookie("atat", session_id)
return self.redirect("/home")
@tornado.gen.coroutine
def _get_user_permissions(self, user_id):
user_repo = Users(self.db_session)
user = user_repo.get_or_create(user_id, atat_role_name="developer")
return user
def get_current_user(self):
cookie = self.get_secure_cookie("atat")
if cookie:
try:
session = self.application.sessions.get_session(cookie)
except SessionNotFoundError:
self.clear_cookie("atat")
return None
else:
return None
return session["user"]

0
atst/handlers/__init__.py
View File

62
atst/handlers/dev.py
View File

@ -1,62 +0,0 @@
import tornado.gen
from atst.handler import BaseHandler
from atst.domain.users import Users
_DEV_USERS = {
"sam": {
"id": "164497f6-c1ea-4f42-a5ef-101da278c012",
"first_name": "Sam",
"last_name": "Seeceepio",
"atat_role": "ccpo",
},
"amanda": {
"id": "cce17030-4109-4719-b958-ed109dbb87c8",
"first_name": "Amanda",
"last_name": "Adamson",
"atat_role": "default",
},
"brandon": {
"id": "66ebf7b8-cbf0-4ed8-a102-5f105330df75",
"first_name": "Brandon",
"last_name": "Buchannan",
"atat_role": "default",
},
"christina": {
"id": "7707b9f2-5945-49ae-967a-be65baa88baf",
"first_name": "Christina",
"last_name": "Collins",
"atat_role": "default",
},
"dominick": {
"id": "6978ac0c-442a-46aa-a0c3-ff17b5ec2a8c",
"first_name": "Dominick",
"last_name": "Domingo",
"atat_role": "default",
},
"erica": {
"id": "596fd001-bb1d-4adf-87d8-fa2312e882de",
"first_name": "Erica",
"last_name": "Eichner",
"atat_role": "default",
},
}
class Dev(BaseHandler):
def initialize(self, action, sessions, db_session):
self.db_session = db_session
self.action = action
self.sessions = sessions
self.users_repo = Users(db_session)
@tornado.gen.coroutine
def get(self):
role = self.get_argument("username", "amanda")
user = _DEV_USERS[role]
yield self._set_user_permissions(user["id"], user["atat_role"])
yield self.login(user)
@tornado.gen.coroutine
def _set_user_permissions(self, user_id, role):
return self.users_repo.get_or_create(user_id, atat_role_name=role)

38
atst/handlers/login_redirect.py
View File

@ -1,38 +0,0 @@
import tornado
from atst.handler import BaseHandler
class LoginRedirect(BaseHandler):
def initialize(self, authnid_client, sessions, db_session):
self.db_session = db_session
self.authnid_client = authnid_client
self.sessions = sessions
@tornado.gen.coroutine
def get(self):
token = self.get_query_argument("bearer-token")
if token:
user = yield self._fetch_user_info(token)
if user:
yield self.login(user)
else:
self.write_error(401)
url = self.get_login_url()
self.redirect(url)
@tornado.gen.coroutine
def _fetch_user_info(self, token):
try:
response = yield self.authnid_client.post(
"/validate", json={"token": token}
)
if response.code == 200:
return response.json["user"]
except tornado.httpclient.HTTPError as error:
if error.response.code == 401:
return None
else:
raise error

11
atst/handlers/main.py
View File

@ -1,11 +0,0 @@
import tornado
from atst.handler import BaseHandler
class Main(BaseHandler):
def initialize(self, page):
self.page = page
@tornado.web.authenticated
def get(self):
self.render("%s.html.to" % self.page, page=self.page)

44
atst/handlers/request.py
View File

@ -1,44 +0,0 @@
import tornado
import pendulum
from atst.handler import BaseHandler
from atst.domain.requests import Requests
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"]),
}
class Request(BaseHandler):
def initialize(self, page, db_session):
self.page = page
self.db_session = db_session
self.requests = Requests(db_session)
@tornado.web.authenticated
@tornado.gen.coroutine
def get(self):
user = self.get_current_user()
requests = yield self.fetch_requests(user)
mapped_requests = [map_request(user, request) for request in requests]
self.render("requests.html.to", page=self.page, requests=mapped_requests)
@tornado.gen.coroutine
def fetch_requests(self, user):
requests = []
if "review_and_approve_jedi_workspace_request" in user["atat_permissions"]:
requests = self.requests.get_many()
else:
requests = self.requests.get_many(creator_id=user["id"])
return requests

64
atst/handlers/request_financial_verification.py
View File

@ -1,64 +0,0 @@
import tornado
from atst.handler import BaseHandler
from atst.forms.financial import FinancialForm
from atst.domain.requests import Requests
from atst.domain.pe_numbers import PENumbers
class RequestFinancialVerification(BaseHandler):
def initialize(self, page, db_session):
self.page = page
self.requests_repo = Requests(db_session)
self.pe_numbers_repo = PENumbers(db_session)
def get_existing_request(self, request_id):
return self.requests_repo.get(request_id)
@tornado.web.authenticated
@tornado.gen.coroutine
def get(self, request_id=None):
existing_request = self.get_existing_request(request_id)
form = FinancialForm(data=existing_request.body.get("financial_verification"))
self.render(
"requests/financial_verification.html.to",
page=self.page,
f=form,
request_id=request_id,
)
@tornado.gen.coroutine
def update_request(self, request_id, form_data):
request_data = {
"creator_id": self.current_user["id"],
"request": {"financial_verification": form_data},
}
return self.requests_repo.update(request_id, request_data)
@tornado.web.authenticated
@tornado.gen.coroutine
def post(self, request_id=None):
self.check_xsrf_cookie()
post_data = self.request.arguments
existing_request = self.get_existing_request(request_id)
form = FinancialForm(post_data)
rerender_args = dict(request_id=request_id, f=form)
if form.validate():
yield self.update_request(request_id, form.data)
# pylint: disable=E1121
valid = yield form.perform_extra_validation(
existing_request.body.get("financial_verification"),
self.pe_numbers_repo,
)
if valid:
self.redirect(
self.application.default_router.reverse_url(
"financial_verification_submitted"
)
)
else:
self.render("requests/financial_verification.html.to", **rerender_args)
else:
self.render("requests/financial_verification.html.to", **rerender_args)

228
atst/handlers/request_new.py
View File

@ -1,228 +0,0 @@
import tornado
from collections import defaultdict
from atst.handler import BaseHandler
from atst.forms.request import RequestForm
from atst.forms.org import OrgForm
from atst.forms.poc import POCForm
from atst.forms.review import ReviewForm
from atst.domain.requests import Requests
from atst.domain.pe_numbers import PENumbers
class RequestNew(BaseHandler):
def initialize(self, page, db_session):
self.page = page
self.requests_repo = Requests(db_session)
self.pe_numbers_repo = PENumbers(db_session)
def get_existing_request(self, request_id):
if request_id is None:
return None
request = self.requests_repo.get(request_id)
return request
@tornado.web.authenticated
@tornado.gen.coroutine
def post(self, screen=1, request_id=None):
self.check_xsrf_cookie()
screen = int(screen)
post_data = self.request.arguments
current_user = self.get_current_user()
existing_request = self.get_existing_request(request_id)
jedi_flow = JEDIRequestFlow(
self.requests_repo,
self.pe_numbers_repo,
screen,
post_data=post_data,
request_id=request_id,
current_user=current_user,
existing_request=existing_request,
)
rerender_args = dict(
f=jedi_flow.form,
data=post_data,
page=self.page,
screens=jedi_flow.screens,
current=screen,
next_screen=jedi_flow.next_screen,
request_id=jedi_flow.request_id,
)
if jedi_flow.validate():
jedi_flow.create_or_update_request()
valid = yield jedi_flow.validate_warnings()
if valid:
if jedi_flow.next_screen > len(jedi_flow.screens):
where = "/requests"
else:
where = self.application.default_router.reverse_url(
"request_form_update",
jedi_flow.next_screen,
jedi_flow.request_id,
)
self.redirect(where)
else:
self.render("requests/screen-%d.html.to" % int(screen), **rerender_args)
else:
self.render("requests/screen-%d.html.to" % int(screen), **rerender_args)
@tornado.web.authenticated
@tornado.gen.coroutine
def get(self, screen=1, request_id=None):
screen = int(screen)
request = None
if request_id:
request = self.requests_repo.get(request_id)
jedi_flow = JEDIRequestFlow(
self.requests_repo,
self.pe_numbers_repo,
screen,
request,
request_id=request_id,
)
self.render(
"requests/screen-%d.html.to" % int(screen),
f=jedi_flow.form,
data=jedi_flow.current_step_data,
page=self.page,
screens=jedi_flow.screens,
current=screen,
next_screen=screen + 1,
request_id=request_id,
can_submit=jedi_flow.can_submit,
)
class JEDIRequestFlow(object):
def __init__(
self,
pe_numbers_repo,
current_step,
request=None,
post_data=None,
request_id=None,
current_user=None,
existing_request=None,
):
self.pe_numbers_repo = pe_numbers_repo
self.requests_repo = requests_repo
self.pe_numbers_repo = pe_numbers_repo
self.current_step = current_step
self.request = request
self.post_data = post_data
self.is_post = self.post_data is not None
self.request_id = request_id
self.form = self._form()
self.current_user = current_user
self.existing_request = existing_request
def _form(self):
if self.is_post:
return self.form_class()(self.post_data)
elif self.request:
return self.form_class()(data=self.current_step_data)
else:
return self.form_class()()
def validate(self):
return self.form.validate()
@tornado.gen.coroutine
def validate_warnings(self):
existing_request_data = (
self.existing_request and self.existing_request.body.get(self.form_section)
) or None
valid = yield self.form.perform_extra_validation(
existing_request_data, self.pe_numbers_repo
)
return valid
@property
def current_screen(self):
return self.screens[self.current_step - 1]
@property
def form_section(self):
return self.current_screen["section"]
def form_class(self):
return self.current_screen["form"]
@property
def current_step_data(self):
data = {}
if self.is_post:
data = self.post_data
if self.request:
if self.form_section == "review_submit":
data = self.request.body
else:
data = self.request.body.get(self.form_section, {})
return defaultdict(lambda: defaultdict(lambda: "Input required"), data)
@property
def can_submit(self):
return self.request and self.request.status != "incomplete"
@property
def next_screen(self):
return self.current_step + 1
@property
def screens(self):
return [
{
"title": "Details of Use",
"section": "details_of_use",
"form": RequestForm,
"subitems": [
{
"title": "Overall request details",
"id": "overall-request-details",
},
{"title": "Cloud Resources", "id": "cloud-resources"},
{"title": "Support Staff", "id": "support-staff"},
],
"show": True,
},
{
"title": "Information About You",
"section": "information_about_you",
"form": OrgForm,
"show": True,
},
{
"title": "Primary Point of Contact",
"section": "primary_poc",
"form": POCForm,
"show": True,
},
{
"title": "Review & Submit",
"section": "review_submit",
"form": ReviewForm,
"show": True,
},
]
def create_or_update_request(self):
request_data = {self.form_section: self.form.data}
if self.request_id:
self.requests_repo.update(self.request_id, request_data)
else:
request = self.requests_repo.create(self.current_user["id"], request_data)
self.request_id = request.id

20
atst/handlers/request_submit.py
View File

@ -1,20 +0,0 @@
import tornado
from atst.handler import BaseHandler
from atst.domain.requests import Requests
class RequestsSubmit(BaseHandler):
def initialize(self, db_session):
self.db_session = db_session
self.requests_repo = Requests(db_session)
@tornado.web.authenticated
@tornado.gen.coroutine
def post(self, request_id):
request = self.requests_repo.get(request_id)
request = yield self.requests_repo.submit(request)
if request.status == "approved":
self.redirect("/requests?modal=True")
else:
self.redirect("/requests")

9
atst/handlers/root.py
View File

@ -1,9 +0,0 @@
from atst.handler import BaseHandler
class Root(BaseHandler):
def initialize(self, page):
self.page = page
def get(self):
self.render("%s.html.to" % self.page, page=self.page)

17
atst/handlers/workspace.py
View File

@ -1,17 +0,0 @@
import tornado
from atst.handler import BaseHandler
from atst.domain.workspaces import Projects
class Workspace(BaseHandler):
def initialize(self):
self.projects_repo = Projects()
@tornado.web.authenticated
@tornado.gen.coroutine
def get(self, workspace_id):
projects = self.projects_repo.get_many(workspace_id)
self.render(
"workspace_projects.html.to", workspace_id=workspace_id, projects=projects
)

17
atst/handlers/workspace_members.py
View File

@ -1,17 +0,0 @@
import tornado
from atst.handler import BaseHandler
from atst.domain.workspaces import Members
class WorkspaceMembers(BaseHandler):
def initialize(self):
self.members_repo = Members()
@tornado.web.authenticated
@tornado.gen.coroutine
def get(self, workspace_id):
members = self.members_repo.get_many(workspace_id)
self.render(
"workspace_members.html.to", workspace_id=workspace_id, members=members
)

22
atst/handlers/workspaces.py
View File

@ -1,22 +0,0 @@
from atst.handler import BaseHandler
import tornado
mock_workspaces = [
{
"name": "Unclassified IaaS and PaaS for Defense Digital Service (DDS)",
"id": "5966187a-eff9-44c3-aa15-4de7a65ac7ff",
"task_order": {"number": 123456},
"user_count": 23,
}
]
class Workspaces(BaseHandler):
def initialize(self, page, db_session):
self.page = page
self.db_session = db_session
@tornado.gen.coroutine
@tornado.web.authenticated
def get(self):
self.render("workspaces.html.to", page=self.page, workspaces=mock_workspaces)