import flask import flask_sqlalchemy import sqlalchemy import sqlalchemy_utils import flask_login import werkzeug.security import datetime import enum import utils app = flask.Flask(__name__) app.config["SQLALCHEMY_DATABASE_URI"] = 'sqlite:///items.db' app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False app.config["SECRET_KEY"] = "placeholder_a4b5c6d7" db = flask_sqlalchemy.SQLAlchemy(app) login_manager = flask_login.LoginManager() login_manager.init_app(app) login_manager.login_view = "route_signin" class UserRoles(enum.IntEnum): UNDEFINED = 0 READONLY = 10 DEVELOPER = 20 MAINTAINER = 30 ADMIN = 40 @property def text(self): return self.name.title() @classmethod def is_valid(cls, role_number: int): return role_number in cls._value2member_map_ class User(flask_login.UserMixin, db.Model): __tablename__ = "user" id = db.Column(db.Integer, primary_key=True) username = db.Column( sqlalchemy_utils.StringEncryptedType(db.String(64), app.config["SECRET_KEY"]), nullable=False) email = db.Column( sqlalchemy_utils.StringEncryptedType(db.String(120), app.config["SECRET_KEY"]), nullable=False) password_hash = db.Column(db.String(128), nullable=False) user_role_value = db.Column(db.Integer, nullable=False, default=UserRoles.UNDEFINED) reports = db.relationship("TestReport", backref="author") @property def user_role(self): try: return UserRoles(self.user_role_value) except ValueError: return UserRoles.UNDEFINED @property def password(self): raise AttributeError("Cannot read password") @password.setter def password(self, password): self.password_hash = werkzeug.security.generate_password_hash(password) def verify_password(self, password): return werkzeug.security.check_password_hash(self.password_hash, password) class BoardFamily(db.Model): __tablename__ = "board_family" board_family_id = db.Column(db.Integer, primary_key=True) board_family_prefix = db.Column(db.String(16), unique=True, nullable=False) boards = db.relationship("Board", backref="family") class Board(db.Model): __tablename__ = "board" board_id = db.Column(db.Integer, primary_key=True) board_family_id = db.Column(db.Integer, db.ForeignKey("board_family.board_family_id"), nullable=False) board_name = db.Column(db.String(32), unique=True, nullable=False) test_boards = db.relationship("TestBoards", backref="board") class OperatingSystem(db.Model): __tablename__ = "operating_system" os_id = db.Column(db.Integer, primary_key=True) os_name = db.Column(db.String(32), unique=True, nullable=False) test_boards = db.relationship("TestBoards", backref="os") class TestBoards(db.Model): __tablename__ = "test_boards" testboard_id = db.Column(db.Integer, primary_key=True) board_id = db.Column(db.Integer, db.ForeignKey("board.board_id"), nullable=False) os_id = db.Column(db.Integer, db.ForeignKey("operating_system.os_id"), nullable=True) # an OS is not always installed _mac_addr = db.Column(db.String(18), nullable=False) test_boards = db.relationship("TestProduct", backref="testboard") @property def mac_addr(self): return self._mac_addr @mac_addr.setter def mac_addr(self, new_addr): self._mac_addr = new_addr class Project(db.Model): __tablename__ = "project" project_id = db.Column(db.Integer, primary_key=True) project_name = db.Column(db.String(64), unique=True, nullable=False) versions = db.relationship("ProjectVersion", backref="project") class ProjectVersion(db.Model): __tablename__ = "project_version" project_version_id = db.Column(db.Integer, primary_key=True) project_id = db.Column(db.Integer, db.ForeignKey("project.project_id"), nullable=False) project_version_major = db.Column(db.Integer, nullable=False) project_version_minor = db.Column(db.Integer, nullable=False) project_version_micro = db.Column(db.Integer, nullable=False) is_hidden = db.Column(db.Boolean, nullable=False) reports = db.relationship("TestReport", backref="project_version") __table_args__ = ( sqlalchemy.UniqueConstraint('project_id', 'project_version_major', 'project_version_minor', 'project_version_micro', name='_project_version_uc'), ) @property def version_string(self): return f"v{self.project_version_major}.{self.project_version_minor}.{self.project_version_micro}" class TestReport(db.Model): __tablename__ = "test_report" test_report_id = db.Column(db.Integer, primary_key=True) author_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False) project_version_id = db.Column(db.Integer, db.ForeignKey("project_version.project_version_id"), nullable=False) project_version_build = db.Column(db.Integer, nullable=False) timestamp_start = db.Column(db.DateTime, nullable=False) timestamp_end = db.Column(db.DateTime, nullable=False) cached_result_pass = db.Column(db.Integer, nullable=False) cached_result_fail = db.Column(db.Integer, nullable=False) tests = db.relationship("TestProduct", backref="report") class TestProduct(db.Model): __tablename__ = "test_product" test_product_id = db.Column(db.Integer, primary_key=True) test_report_id = db.Column(db.Integer, db.ForeignKey("test_report.test_report_id"), nullable=False) testboard_id = db.Column(db.Integer, db.ForeignKey("test_boards.testboard_id"), nullable=False) timestamp_start = db.Column(db.DateTime, nullable=False) timestamp_end = db.Column(db.DateTime, nullable=False) cached_result_pass = db.Column(db.Integer, nullable=False) cached_result_fail = db.Column(db.Integer, nullable=False) test_items = db.relationship("TestItem", backref="test") class TestItem(db.Model): __tablename__ = "test_item" test_item_id = db.Column(db.Integer, primary_key=True) test_product_id = db.Column(db.Integer, db.ForeignKey("test_product.test_product_id"), nullable=False) test_item_name = db.Column(db.String(128), nullable=False) test_script_version = db.Column(db.String(32), nullable=False) test_item_result = db.Column(db.Boolean, nullable=False) @login_manager.user_loader def load_user(user_id): return User.query.get(int(user_id)) @app.before_request def load_logged_in_user(): user_id = flask.session.get("user_id") if user_id is None: flask.g.user = None else: flask.g.user = User.query.get(user_id) @app.route("/") def route_home(): return flask.render_template("home.html") @app.route("/signout") @flask_login.login_required def route_logout(): flask_login.logout_user() return flask.redirect("/signin") @app.route("/signin", methods=["GET", "POST"]) def route_signin(): if flask.request.method == "GET": return flask.render_template("signin.html") username = flask.request.form.get("username", "") password = flask.request.form.get("password", "") if not username: flask.flash("Username is blank", "error") return flask.redirect("/signin") if not password: flask.flash("Password is blank", "error") return flask.redirect("/signin") if not utils.VALID in ( utils.validation.username(username), utils.validation.email(username)): flask.flash("Username contains illegal characters") return flask.redirect("/signin") user = User.query.filter_by(username=username).first() if not user: user = User.query.filter_by(email=username).first() if not user: flask.flash("Incorrect _Username_ or Password", "error") return flask.render_template("/signin.html") if not werkzeug.security.check_password_hash(user.password_hash, password): flask.flash("Incorrect Username or _Password_", "error") return flask.render_template("/signin.html") flask_login.login_user(user) return flask.redirect("/account") @app.route("/product-setup") @flask_login.login_required def route_product_setup(): families = BoardFamily.query.order_by(BoardFamily.board_family_prefix).all() products_table = [] for family in families: products_table.append( Board.query.filter_by(board_family_id=family.board_family_id).order_by(Board.board_name.desc()).all() ) return flask.render_template("product_setup.html", families=families, products_table=products_table) @app.route("/product-setup/add-family", methods=["POST"]) def route_product_setup_add_family(): if flask_login.current_user.user_role < UserRoles.MAINTAINER: flask.flash("Insufficient privilages", "error") return flask.redirect("/product-setup") family_name = flask.request.form.get("family_name", "").strip() if not family_name: flask.flash("Family name is null", "error") return flask.redirect("/product-setup") if utils.validation.text_numeric_only(family_name, extra_chars="-") == utils.INVALID: flask.flash("Family name contains illegal characters", "error") return flask.redirect("/product-setup") existing_family = BoardFamily.query.filter_by(board_family_prefix=family_name).first() if existing_family: flask.flash("Family name already exists", "error") return flask.redirect("/product-setup") new_family = BoardFamily(board_family_prefix=family_name) db.session.add(new_family) db.session.commit() return flask.redirect("/product-setup") @app.route("/product-setup/add-board", methods=["POST"]) @flask_login.login_required def route_product_setup_add_board(): if flask_login.current_user.user_role < UserRoles.MAINTAINER: flask.flash("Insufficient privilages", "error") return flask.redirect("/product-setup") board_name = flask.request.form.get("board_name", "").strip() board_family_id = flask.request.form.get("board_family_id", default=0, type=int) if not board_name: flask.flash("Board name is null", "error") return flask.redirect("/product-setup") if utils.validation.text_numeric_only(board_name) == utils.INVALID: flask.flash("Board name contains illegal characters", "error") return flask.redirect("/product-setup") if board_family_id == 0: flask.flash("Board family is invalid", "error") return flask.redirect("/product-setup") existing_board = Board.query.filter_by(board_name=board_name).first() if existing_board: flask.flash("Board name already exists", "error") return flask.redirect("/product-setup") new_board = Board(board_family_id=board_family_id, board_name=board_name) db.session.add(new_board) db.session.commit() return flask.redirect("/product-setup") @app.route("/operating-system-setup") @flask_login.login_required def route_operating_system_setup(): operating_systems = OperatingSystem.query.order_by(OperatingSystem.os_name).all() return flask.render_template("operating_system_setup.html", operating_systems=operating_systems) @app.route("/operating-system-setup/add-os", methods=["POST"]) @flask_login.login_required def route_operating_system_setup_add_os(): if flask_login.current_user.user_role < UserRoles.MAINTAINER: flask.flash("Insufficient privilages", "error") return flask.redirect("/operating-system-setup") os_name = flask.request.form.get("os_name", "").strip() if not os_name: flask.flash("OS name is invalid", "error") return flask.redirect("/operating-system-setup") if utils.validation.text_numeric_only(os_name, extra_chars=" -.") == utils.INVALID: flask.flash("OS name contains illegal characters", "error") return flask.redirect("/operating-system-setup") os = OperatingSystem.query.filter_by(os_name=os_name).first() if os is not None: flask.flash(f"OS already exists: {os_name}", "error") return flask.redirect("/operating-system-setup") os = OperatingSystem(os_name=os_name) db.session.add(os) db.session.commit() return flask.redirect("/operating-system-setup") @app.route("/operating-system-setup/delete-os/", methods=["POST"]) @flask_login.login_required def route_operating_system_setup_delete_os(os_id): if flask_login.current_user.user_role < UserRoles.MAINTAINER: flask.flash("Insufficient privilages", "error") return flask.redirect("/operating-system-setup") operating_system = OperatingSystem.query.filter_by(os_id=os_id).first() if operating_system is None: flask.flash("OS not found", "error") return flask.redirect("/operating-system-setup") if len(operating_system.test_boards) != 0: flask.flash("This OS is still in use by a test board", "error") return flask.redirect("/operating-system-setup") db.session.delete(operating_system) db.session.commit() return flask.redirect("/operating-system-setup") @app.route("/test-bench-setup") @flask_login.login_required def route_test_bench_setup(): families = BoardFamily.query.order_by(BoardFamily.board_family_prefix).all() boards = Board.query.order_by(Board.board_name).all() operating_systems = OperatingSystem.query.order_by(OperatingSystem.os_name).all() test_boards = TestBoards.query.all() return flask.render_template("test_bench.html", families=families, boards=boards, operating_systems=operating_systems, test_boards=test_boards) @app.route("/test-bench-setup/add-board", methods=["POST"]) @flask_login.login_required def route_test_bench_setup_add_board(): if flask_login.current_user.user_role < UserRoles.MAINTAINER: flask.flash("Insufficient privilages", "error") return flask.redirect("/test-bench-setup") board_id = flask.request.form.get("board_id", default=0, type=int) mac_addr = utils.sanitise.mac_addr(flask.request.form.get("mac_addr", "").strip()) if board_id == 0: flask.flash("Board ID is invalid", "error") return flask.redirect("/test-bench-setup") if not mac_addr: flask.flash("MAC address is null", "error") return flask.redirect("/test-bench-setup") if utils.validation.mac_addr(mac_addr) != utils.VALID: flask.flash("MAC address is not valid", "error") return flask.redirect("/test-bench-setup") new_test_board = TestBoards(board_id=board_id, mac_addr=mac_addr) db.session.add(new_test_board) db.session.commit() return flask.redirect("/test-bench-setup") @app.route("/test-bench-setup/update-os/", methods=["POST"]) @flask_login.login_required def route_test_bench_setup_update_os(testboard_id): if flask_login.current_user.user_role < UserRoles.MAINTAINER: flask.flash("Insufficient privilages", "error") return flask.redirect("/test-bench-setup") os_id = flask.request.form.get("os_id", default=0, type=int) if testboard_id == 0: flask.flash("Test baord ID is null", "error") return flask.redirect("/test-bench-setup") test_board = TestBoards.query.filter_by(testboard_id=testboard_id).first() if test_board is None: flask.flash("Test board not found", "error") return flask.redirect("/test-bench-setup") # if we are selecting a non-zero OS, we must check it is valid # 0 indicated a null value - where no OS is installed if os_id != 0: os = OperatingSystem.query.filter_by(os_id=os_id).first() if os is None: flask.flash("OS is not found", "error") return flask.redirect("/test-bench-setup") test_board.os_id = os_id db.session.commit() return flask.redirect("/test-bench-setup") @app.route("/test-bench-setup/delete-test-board/", methods=["POST"]) @flask_login.login_required def route_test_bench_setup_delete_test_board(testboard_id): if flask_login.current_user.user_role < UserRoles.MAINTAINER: flask.flash("Insufficient privilages", "error") return flask.redirect("/test-bench-setup") if testboard_id == 0: flask.flash("Test board ID is null", "error") return flask.redirect("/test-bench-setup") test_board = TestBoards.query.filter_by(testboard_id=testboard_id).first() if test_board is None: flask.flash("Test board is not found", "error") return flask.redirect("/test-bench-setup") db.session.delete(test_board) db.session.commit() return flask.redirect("/test-bench-setup") @app.route("/schedule-new-test", methods=["GET", "POST"]) @flask_login.login_required def route_schedule_new_test(): if flask_login.current_user.user_role < UserRoles.DEVELOPER: flask.flash("Insufficient privilages", "error") return flask.redirect("/") if flask.request.method == "GET": operating_systems = OperatingSystem.query.order_by(OperatingSystem.os_name).all() test_boards = TestBoards.query.all() projects = Project.query.all() projects_dict = {} for project in projects: projects_dict[project.project_name] = [] for version in project.versions: if version.is_hidden: continue v_maj = version.project_version_major v_min = version.project_version_minor v_mic = version.project_version_micro ver_str = f"{v_maj}.{v_min}.{v_mic}" projects_dict[project.project_name].append(ver_str) return flask.render_template("schedule_new_test.html", operating_systems=operating_systems, test_boards=test_boards, projects=projects_dict) # handle post request project = utils.sanitise.text_input(flask.request.form.get("proj", "").strip()) version = utils.sanitise.text_input(flask.request.form.get("ver", "").strip()) build_num = flask.request.form.get("build", type=int) test_board_ids = flask.request.form.getlist("test-boards") # load project id project_obj = Project.query.filter_by(project_name=project).first() if project_obj is None: flask.flash("Cannot load the project", "error") return flask.redirect("/schedule-new-test") project_id = project_obj.project_id generic_version_error_msg = "Version provided is not in the correct format" # parse project version if version.count(".") != 2: flask.flash(generic_version_error_msg, "error") return flask.redirect("/schedule-new-test") v_maj, v_min, v_mic = version.split(".") try: v_maj = int(v_maj) v_min = int(v_min) v_mic = int(v_mic) except ValueError: flask.flash(generic_version_error_msg, "error") return flask.redirect("/schedule-new-test") version_obj = ProjectVersion.query.filter_by( project_id=project_id, project_version_major=v_maj, project_version_minor=v_min, project_version_micro=v_mic, ).first() if version_obj is None: flask.flash("Cannot load the provided version", "error") return flask.redirect("/schedule-new-test") project_version_id = version_obj.project_version_id # create TestReport record test_report_obj = TestReport( author_id = flask_login.current_user.id, project_version_id = project_version_id, project_version_build = 4, # placeholder timestamp_start = datetime.datetime.now(), timestamp_end = utils.defaults.unix_epoch(), cached_result_pass = 0, # placeholder cached_result_fail = 0, # placeholder ) db.session.add(test_report_obj) # todo: create TestProduct for each selected board test_products_objs = [] for test_board_id in test_board_ids: board_obj = TestBoards.query.filter_by(testboard_id=test_board_id).first() if board_obj is None: db.session.rollback() flask.flash("Cannot cancel this report", "error") return flask.redirect("/schedule-new-test") test_board_obj = TestProduct( test_report_id = test_report_obj.test_report_id, testboard_id = test_board_id, timestamp_start = utils.defaults.unix_epoch(), timestamp_end = utils.defaults.unix_epoch(), cached_result_pass = 0, cached_result_fail = 0, ) test_products_objs.append(test_board_obj) for test_product in test_products_objs: db.session.add(test_product) db.session.commit() # not fully implemented due to backend relying on internal coporate systems # this is outside the scope of this assignment return flask.redirect("/active-tests") @app.route("/active-tests") @flask_login.login_required def route_active_tests(): active_reports = TestReport.query.filter_by(timestamp_end=utils.defaults.unix_epoch()).all() return flask.render_template("active-tests.html", active_reports=active_reports) @app.route("/active-tests/delete/", methods=["POST"]) @flask_login.login_required def route_active_tests_delete(report_id): if flask_login.current_user.user_role < UserRoles.DEVELOPER: flask.flash("Insufficient privilages", "error") return flask.redirect("/active-tests") report = TestReport.query.filter_by(test_report_id=report_id, timestamp_end=utils.defaults.unix_epoch()).first() if report is None: flask.flash("Cannot cancel this report", "error") return flask.redirect("/active-tests") # todo: delete child nodes first child_test_products = TestProduct.query.filter_by(test_report_id=report.test_report_id).all() for test_product in child_test_products: db.session.delete(test_product) db.session.delete(report) db.session.commit() return flask.redirect("/active-tests") @app.route("/active-tests/") @flask_login.login_required def route_active_tests_products(report_id): test_products = TestProduct.query.filter_by(test_report_id=report_id).all() if len(test_products) == 0: flask.flash(f"Invalid test report ID: {report_id}") return flask.redirect("/active-tests") return flask.render_template("active-tests-products.html", report_id=report_id, test_products=test_products) @app.route("/active-tests/delete-product-test/", methods=["POST"]) @flask_login.login_required def route_active_tests_delete_product_test(product_test_id): if flask_login.current_user.user_role < UserRoles.MAINTAINER: flask.flash("Insufficient privilages", "error") return flask.redirect(f"/active-tests/{product_test_id}") test_product = TestProduct.query.filter_by(test_product_id=product_test_id).first() if test_product is None: flask.flash("Cannot load record from database") return flask.redirect(f"/active-tests/{product_test_id}") test_report_id = test_product.report.test_report_id db.session.delete(test_product) db.session.commit() return flask.redirect(f"/active-tests/{test_report_id}") @app.route("/test-reports") @flask_login.login_required def route_test_reports(): return flask.render_template("test_reports.html") @app.route("/projects") @flask_login.login_required def route_projects(): all_projects = Project.query.order_by(Project.project_name).all() return flask.render_template("projects.html", projects=all_projects) @app.route("/projects/add-project", methods=["POST"]) @flask_login.login_required def route_add_project(): if flask_login.current_user.user_role < UserRoles.MAINTAINER: flask.flash("Insufficient privilages", "error") return flask.redirect("/projects") project_name = flask.request.form.get("project_name", "").strip() if not project_name: flask.flash("Project name is null", "error") return flask.redirect("/projects") if utils.validation.text_numeric_only(project_name, extra_chars=" -") == utils.INVALID: flask.flash("Project name contains illegal characters", "error") return flask.redirect("/projects") existing_project = Project.query.filter_by(project_name=project_name).first() if existing_project: flask.flash("Project with that name already exists", "error") return flask.redirect("/projects") new_project = Project(project_name=project_name) db.session.add(new_project) db.session.commit() return flask.redirect("/projects") @app.route("/projects/add-version", methods=["POST"]) @flask_login.login_required def route_add_project_version(): if flask_login.current_user.user_role < UserRoles.MAINTAINER: flask.flash("Insufficient privilages", "error") return flask.redirect("/projects") project_id = flask.request.form.get("project_id", type=int) version_major = flask.request.form.get("major", type=int) version_minor = flask.request.form.get("minor", type=int) version_micro = flask.request.form.get("micro", type=int) if not project_id: flask.flash("Project ID is not defined", "error") return flask.redirect("/projects") if None in (version_major, version_minor, version_micro): flask.flash("Not all version levels are defined (major minor micro)", "error") return flask.redirect("/projects") existing_version = ProjectVersion.query.filter_by( project_id=project_id, project_version_major=version_major, project_version_minor=version_minor, project_version_micro=version_micro ).first() if existing_version: flask.flash("Version already exists for this project", "error") return flask.redirect("/projects") new_version = ProjectVersion( project_id=project_id, project_version_major=version_major, project_version_minor=version_minor, project_version_micro=version_micro, is_hidden=False ) db.session.add(new_version) db.session.commit() return flask.redirect("/projects") @app.route("/projects/toggle-version-hidden/", methods=["POST"]) @flask_login.login_required def route_toggle_version_hidden(project_version_id): if flask_login.current_user.user_role < UserRoles.MAINTAINER: flask.flash("Insufficient privilages", "error") return flask.redirect("/projects") version = ProjectVersion.query.filter_by(project_version_id=project_version_id).first() if not version: flask.flash("Version cannot be loaded from database", "error") return flask.redirect("/projects") version.is_hidden = not version.is_hidden db.session.commit() return flask.redirect("/projects") @app.route("/account") @flask_login.login_required def route_accounts(): users = User.query.all() return flask.render_template("account.html", users=users, UserRoles=UserRoles) @app.route("/account/new", methods=["GET", "POST"]) @flask_login.login_required def route_account_new(): if flask_login.current_user.user_role < UserRoles.ADMIN: flask.flash("Insufficient privilages", "error") return flask.redirect("/account") if flask.request.method == "POST": username = flask.request.form.get("username") email = flask.request.form.get("email") password = flask.request.form.get("password") usertype = flask.request.form.get("user-type", type=int) if None in (username, email, password): flask.flash("fields cannot be blank", "error") return flask.redirect("/account/new") if utils.INVALID == utils.validation.username(username): flask.flash("Username contains illegal characters", "error") return flask.redirect("/account/new") if utils.INVALID == utils.validation.email(email): flask.flash("Email contains illegal characters", "error") return flask.redirect("/account/new") users_via_username = User.query.filter_by(username=username).first() users_via_email = User.query.filter_by(email=email).first() if (users_via_username is not None) or (users_via_email is not None): flask.flash("That user already exists", "error") return flask.redirect("/account/new") if not UserRoles.is_valid(usertype): flask.flash("Invalid user role") return flask.redirect("/account/new") # todo actually create the new user user = User( username=username, email=email, password=password, user_role_value=usertype ) db.session.add(user) db.session.commit() return flask.redirect("/account") return flask.render_template("account-new.html", UserRoles=UserRoles) @app.route("/account/update", methods=["POST"]) @flask_login.login_required def route_account_update(): if flask_login.current_user.user_role < UserRoles.ADMIN: flask.flash("Insufficient privilages", "error") return flask.redirect("/account") user_id = flask.request.form.get("user-id", type=int) userrole = flask.request.form.get("user-type", type=int) user_obj = User.query.filter_by(id=user_id).first() if user_obj is None: flask.flash("Cannot load that user", "error") return flask.redirect("/account") if not UserRoles.is_valid(userrole): flask.flash("User role is not valid", "error") return flask.redirect("/account") if user_obj.user_role_value == userrole: flask.flash("User role has not been changed", "error") return flask.redirect("/account") user_obj.user_role_value = userrole db.session.commit() with app.app_context(): db.create_all() # debug create initial account if not User.query.first(): initial_user = User( username="admin", email="admin@localhost", password="admin", user_role_value=UserRoles.ADMIN ) db.session.add(initial_user) db.session.commit() if __name__ == "__main__": app.run(host="0.0.0.0", port=8080)