def submit_assessment()

in microservices/assessment_service/src/services/submitted_assessment.py [0:0]

• 95 lines of code
• 24 McCabe index (conditional complexity)

def submit_assessment(submitted_assessment_dict, header):
  """Helper function to handle the submit assessment flow"""
  assessor_id = None
  pass_status = None
  result = None
  session_data = None
  items = {}

  assessment_id = submitted_assessment_dict["assessment_id"]
  assessment = Assessment.find_by_uuid(assessment_id)
  is_autogradable = assessment.is_autogradable
  max_attempts = assessment.max_attempts
  attempt_no = submitted_assessment_dict["attempt_no"]
  submission_gcs_paths = submitted_assessment_dict.get(
    "submission_gcs_paths", [])
  valid_gcs_path = bool(submission_gcs_paths)
  #pylint: disable=chained-comparison
  if max_attempts is not None and max_attempts > 0 and \
    attempt_no > max_attempts:
    raise PreconditionFailedError(
      f"Allowed number of attempts exceeded ({attempt_no}/{max_attempts})")

  learner_id = submitted_assessment_dict["learner_id"]
  Learner.find_by_uuid(learner_id)
  user = User.find_by_user_type_ref(learner_id)

  if is_autogradable:
    activity_id = assessment.assessment_reference
    if activity_id is not None and activity_id != {}:
      activity_id = activity_id.get("activity_template_id", None)

    learnosity_req_body = {
        "user_id": user.id,
        "session_id": submitted_assessment_dict["learner_session_id"],
        "activity_id": activity_id
    }
    if USE_LEARNOSITY_SECRET:
      session_data = fetch_response(**learnosity_req_body)
      if assessment.metadata is not None:
        items = assessment.metadata.get("items", {})
      items = update_item_responses(items, session_data)
    # Autogradable assessments are always passed

    # Load Sample data to run e2e for pretests
    if not USE_LEARNOSITY_SECRET and assessment.type == "pretest":
      with open(LEARNOSITY_DATA, encoding="utf-8") as data:
        session_data = json.load(data)
      with open(ITEM_REFERENCES, encoding="utf-8") as data:
        items = json.load(data)
    pass_status = True
    status = "completed"
    result = "Pass"
  else:
    status = "evaluation_pending"
    if submission_gcs_paths:
      for path in submission_gcs_paths:
        path = f"gs://{CONTENT_SERVING_BUCKET}/{path}"
        valid_current_gcs_path = is_valid_path(path)
        valid_gcs_path = valid_gcs_path and valid_current_gcs_path
        if not valid_current_gcs_path:
          raise ValidationError(
            "The following GCS Path for the assessment submission does" +\
              f" not exist: {path}")
    # assign assessor only if assessment is human graded and is not an srl
    # or pretest
    if assessment.type not in ["srl", "static_srl", "cognitive_wrapper",
                               "pretest"]:
      discipline = traverse_up(assessment, "assessments", "discipline")
      pathway_id = discipline.get_fields()[
        "uuid"] if discipline is not None else ""
      assessor_id = assessor_handler(header, pathway_id)

  # create document for submitted_assessment
  submitted_assessment_dict = {
    **submitted_assessment_dict, "type": assessment.type,
    "pass_status": pass_status,
    "status": status,
    "result": result,
    "attempt_no": attempt_no,
    "is_autogradable": is_autogradable,
    "learner_session_data": session_data,
    "metadata": {
      "tag_info": items
    }
  }
  new_submitted_assessment = SubmittedAssessment()
  new_submitted_assessment = new_submitted_assessment.from_dict(
      submitted_assessment_dict)
  new_submitted_assessment.assessor_id = assessor_id
  new_submitted_assessment.uuid = ""
  new_submitted_assessment.save()
  new_submitted_assessment.uuid = new_submitted_assessment.id
  if valid_gcs_path:
    attached_files_path = attach_files_to_assessment_submission(
            learner_id, assessment_id, new_submitted_assessment.id,
            submission_gcs_paths, CONTENT_SERVING_BUCKET)
    new_submitted_assessment.submission_gcs_paths = attached_files_path
  elif submission_gcs_paths:
    raise ValidationError(
          "Some of the GCS Path for the assessment submission does not exist")
  new_submitted_assessment.timer_start_time = \
      new_submitted_assessment.created_time
  new_submitted_assessment.update()

  submitted_assessment_fields = new_submitted_assessment.get_fields(
      reformat_datetime=True)
  submitted_assessment_fields["timer_start_time"] = str(
      submitted_assessment_fields["timer_start_time"])
  return submitted_assessment_fields