def test_flow_desc_access_rule_has_change_answer_perm(self):
     flow_desc_dict = self.get_hacked_flow_desc(as_dict=True)
     rules = flow_desc_dict["rules"]
     rules.access = [dict_to_struct(
         {"permissions": ["submit_answer", "change_answer"]})]
     flow_desc = dict_to_struct(flow_desc_dict)
     self.assertTrue(analytics.is_flow_multiple_submit(flow_desc))

def get_yaml_from_repo_side_effect(repo, full_name, commit_sha, cached=True):
    if full_name == events_file:
        return dict_to_struct(
            {"event_kinds": dict_to_struct({
                "lecture": dict_to_struct({
                    "title": "Lecture {nr}",
                    "color": "blue"
                })}),
                "events": dict_to_struct({
                    "lecture 1": dict_to_struct({
                        "title": "l1"})
                })})
    else:
        return get_yaml_from_repo(repo, full_name, commit_sha, cached)

 def test_parse_matcher_instance_is_struct_no_type_error(self):
     s = dict_to_struct(
         {"value": "20.1"})
     with self.assertRaises(ValidationError) as cm:
         parse_matcher(None, "some where", s)
     self.assertIn("some where: matcher must supply 'type'",
                   str(cm.exception))

def get_yaml_from_repo(repo, full_name, commit_sha, cached=True):
    """Return decoded, struct-ified YAML data structure from
    the given file in *repo* at *commit_sha*.

    See :class:`relate.utils.Struct` for more on
    struct-ification.
    """

    if cached:
        cache_key = "%%%2".join((repo.controldir(), full_name, commit_sha))

        import django.core.cache as cache

        def_cache = cache.caches["default"]
        result = def_cache.get(cache_key)
        if result is not None:
            return result

    result = dict_to_struct(
        load_yaml(expand_yaml_macros(repo, commit_sha, get_repo_blob(repo, full_name, commit_sha).data))
    )

    if cached:
        def_cache.add(cache_key, result, None)

    return result

def get_flow_rules(flow_desc, kind, participation, flow_id, now_datetime,
        consider_exceptions=True, default_rules_desc=[]):
    if (not hasattr(flow_desc, "rules")
            or not hasattr(flow_desc.rules, kind)):
        rules = default_rules_desc[:]
    else:
        rules = getattr(flow_desc.rules, kind)[:]

    from course.models import FlowRuleException
    if consider_exceptions:
        for exc in (
                FlowRuleException.objects
                .filter(
                    participation=participation,
                    active=True,
                    kind=kind,
                    flow_id=flow_id)
                # rules created first will get inserted first, and show up last
                .order_by("creation_time")):

            if exc.expiration is not None and now_datetime > exc.expiration:
                continue

            from relate.utils import dict_to_struct
            rules.insert(0, dict_to_struct(exc.rule))

    return rules

    def test_choice_not_stringifiable(self):
        expected_page_error = (
            "choice 10: unable to convert to string")

        class BadChoice(object):
            def __str__(self):
                raise Exception

        from relate.utils import dict_to_struct
        fake_page_desc = dict_to_struct(
            {'type': 'SurveyChoiceQuestion', 'id': 'age_group_with_comment',
             'answer_comment': 'this is a survey question',
             'prompt': '\n# Age\n\nHow old are you?\n',
             'choices': [
                 '0-10 years', '11-20 years', '21-30 years', '31-40 years',
                 '41-50 years', '51-60 years', '61-70 years', '71-80 years',
                 '81-90 years', BadChoice()],
             '_field_names': ['type', 'id', 'answer_comment',
                              'prompt', 'choices']}
        )

        with mock.patch("relate.utils.dict_to_struct") as mock_dict_to_struct:
            mock_dict_to_struct.return_value = fake_page_desc

            markdown = SURVEY_CHOICE_QUESTION_MARKDOWN

            resp = (
                self.get_page_sandbox_preview_response(markdown))
            self.assertEqual(resp.status_code, 200)
            self.assertSandboxNotHasValidPage(resp)
            self.assertResponseContextContains(resp, PAGE_ERRORS,
                                               expected_page_error)

    def validate(self, new_page_source):
        from relate.utils import dict_to_struct
        import yaml

        try:
            page_desc = dict_to_struct(yaml.safe_load(new_page_source))

            from course.validation import (
                    validate_flow_page, ValidationContext)
            vctx = ValidationContext(
                    # FIXME
                    repo=None,
                    commit_sha=None)

            validate_flow_page(vctx, "submitted page", page_desc)

            if page_desc.type != self.validator_desc.page_type:
                raise ValidationError(ugettext("page must be of type '%s'")
                        % self.validator_desc.page_type)

        except Exception:
            tp, e, _ = sys.exc_info()

            raise forms.ValidationError("%(err_type)s: %(err_str)s"
                    % {"err_type": tp.__name__, "err_str": str(e)})

def check_attributes_yml(vctx, repo, path, tree):
    try:
        _, attr_blob_sha = tree[".attributes.yml"]
    except KeyError:
        # no .attributes.yml here
        pass
    else:
        from relate.utils import dict_to_struct
        from yaml import load as load_yaml

        att_yml = dict_to_struct(load_yaml(repo[attr_blob_sha].data))

        loc = path + "/" + ".attributes.yml"
        validate_struct(
                vctx, loc, att_yml,
                required_attrs=[],
                allowed_attrs=[
                    ("public", list),
                ])

        if hasattr(att_yml, "public"):
            for i, l in enumerate(att_yml.public):
                if not isinstance(l, (str, unicode)):
                    raise ValidationError(
                            "%s: entry %d in 'public' is not a string"
                            % (loc, i+1))

    import stat
    for entry in tree.items():
        if stat.S_ISDIR(entry.mode):
            _, blob_sha = tree[entry.path]
            subtree = repo[blob_sha]
            check_attributes_yml(vctx, repo, path+"/"+entry.path, subtree)

def check_attributes_yml(vctx, repo, path, tree):
    try:
        _, attr_blob_sha = tree[".attributes.yml"]
    except KeyError:
        # no .attributes.yml here
        pass
    else:
        from relate.utils import dict_to_struct
        from yaml import load as load_yaml

        att_yml = dict_to_struct(load_yaml(repo[attr_blob_sha].data))

        loc = path + "/" + ".attributes.yml"
        validate_struct(vctx, loc, att_yml, required_attrs=[], allowed_attrs=[("public", list), ("in_exam", list)])

        for access_kind in ["public", "in_exam"]:
            if hasattr(att_yml, access_kind):
                for i, l in enumerate(att_yml.public):
                    if not isinstance(l, six.string_types):
                        raise ValidationError("%s: entry %d in '%s' is not a string" % (loc, i + 1, access_kind))

    import stat

    for entry in tree.items():
        if stat.S_ISDIR(entry.mode):
            _, blob_sha = tree[entry.path]
            subtree = repo[blob_sha]
            check_attributes_yml(vctx, repo, path + "/" + entry.path.decode("utf-8"), subtree)

def get_yaml_from_repo(repo, full_name, commit_sha, cached=True):
    """Return decoded, struct-ified YAML data structure from
    the given file in *repo* at *commit_sha*.

    See :class:`relate.utils.Struct` for more on
    struct-ification.
    """

    if cached:
        from six.moves.urllib.parse import quote_plus
        cache_key = "%%%2".join(
                (quote_plus(repo.controldir()), quote_plus(full_name),
                    commit_sha.decode()))

        import django.core.cache as cache
        def_cache = cache.caches["default"]
        result = None
        # Memcache is apparently limited to 250 characters.
        if len(cache_key) < 240:
            result = def_cache.get(cache_key)
        if result is not None:
            return result

    expanded = expand_yaml_macros(
            repo, commit_sha,
            get_repo_blob(repo, full_name, commit_sha).data)

    result = dict_to_struct(load_yaml(expanded))

    if cached:
        def_cache.add(cache_key, result, None)

    return result

 def test_parse_matcher_instance_is_struct(self):
     s = dict_to_struct(
         {"type": "float",
          "value": "20.1",
          })
     result = parse_matcher(None, "", s)
     self.assertTrue(isinstance(result, FloatMatcher))
     self.assertEqual(result.correct_answer_text(), "20.1")

 def test_float_matcher_value_zero_rtol_zero_error(self):
     expected_error_msg = "'rtol' not allowed when 'value' is zero"
     with self.assertRaises(ValidationError) as cm:
         FloatMatcher(None, "",
                      dict_to_struct(
                          {"type": "float",
                           "value": "0",
                           "rtol": "0"}))
     self.assertIn(expected_error_msg, str(cm.exception))

 def test_float_matcher_neither_atol_nor_rtol_present_warning(self):
     mock_vctx = mock.MagicMock()
     expected_warning = ("Float match should have either rtol or atol--"
                         "otherwise it will match any number")
     FloatMatcher(mock_vctx, "some where",
                  dict_to_struct(
                      {"type": "float",
                       "value": "1"}))
     self.assertIn(expected_warning, mock_vctx.add_warning.call_args[0])

 def test_float_matcher_atol_error(self):
     expected_error_msg = "'atol' does not provide a valid float literal"
     with self.assertRaises(ValidationError) as cm:
         FloatMatcher(None, "",
                      dict_to_struct(
                          {"type": "float",
                           "value": "1",
                           "atol": "abcd"}))
     self.assertIn(expected_error_msg, str(cm.exception))

    def test_float_matcher_value_zero_atol_not_present_warning(self):
        mock_vctx = mock.MagicMock()
        expected_warning = ("Float match for 'value' zero should have "
                            "atol--otherwise it will match any number")
        FloatMatcher(mock_vctx, "some where",
                     dict_to_struct(
                         {"type": "float",
                          "value": "0"}))

        self.assertIn(expected_warning, mock_vctx.add_warning.call_args[0])

    def test_float_matcher_grade_neither_rtol_nor_atol(self):
        matcher = FloatMatcher(None, "",
                               dict_to_struct(
                                   {"type": "float",
                                    "value": "20.1",
                                    }))
        self.assertEqual(matcher.grade(""), 0)
        self.assertEqual(matcher.grade("abcd"), 0)

        self.assertEqual(matcher.grade(20000), 1)
        self.assertEqual(matcher.grade(-2), 1)

    def test_float_matcher_grade_inf(self):
        matcher = FloatMatcher(None, "",
                               dict_to_struct(
                                   {"type": "float",
                                    "value": "inf",
                                    "rtol": 0.01
                                    }))

        self.assertEqual(matcher.grade(float("nan")), 0)
        self.assertEqual(matcher.grade(float("inf")), 1)
        self.assertEqual(matcher.grade(float("20.5")), 0)

def get_session_grading_rule(session, role, flow_desc, now_datetime):
    flow_desc_rules = getattr(flow_desc, "rules", None)

    from relate.utils import dict_to_struct
    rules = get_flow_rules(flow_desc, flow_rule_kind.grading,
            session.participation, session.flow_id, now_datetime,
            default_rules_desc=[
                dict_to_struct(dict(
                    generates_grade=False,
                    ))])

    for rule in rules:
        if hasattr(rule, "if_has_role"):
            if role not in rule.if_has_role:
                continue

        if hasattr(rule, "if_has_tag"):
            if session.access_rules_tag != rule.if_has_tag:
                continue

        if hasattr(rule, "if_completed_before"):
            ds = parse_date_spec(session.course, rule.if_completed_before)
            if session.in_progress and now_datetime > ds:
                continue
            if not session.in_progress and session.completion_time > ds:
                continue

        due = parse_date_spec(session.course, getattr(rule, "due", None))
        if due is not None:
            assert due.tzinfo is not None

        generates_grade = getattr(rule, "generates_grade", True)

        grade_identifier = None
        grade_aggregation_strategy = None
        if flow_desc_rules is not None:
            grade_identifier = flow_desc_rules.grade_identifier
            grade_aggregation_strategy = getattr(
                    flow_desc_rules, "grade_aggregation_strategy", None)

        return FlowSessionGradingRule(
                grade_identifier=grade_identifier,
                grade_aggregation_strategy=grade_aggregation_strategy,
                due=due,
                generates_grade=generates_grade,
                description=getattr(rule, "description", None),
                credit_percent=getattr(rule, "credit_percent", 100),
                use_last_activity_as_completion_time=getattr(
                    rule, "use_last_activity_as_completion_time", False),
                )

    raise RuntimeError(_("grading rule determination was unable to find "
            "a grading rule"))

    def test_float_matcher_validate(self):
        matcher = FloatMatcher(None, "",
                               dict_to_struct(
                                   {"type": "float",
                                    "value": "1",
                                    "atol": 0.01
                                    }))
        matcher.validate(1.1)

        expected_error_msg = "TypeError: can\'t convert expression to float"
        with self.assertRaises(forms.ValidationError) as cm:
            matcher.validate("abcd")
        self.assertIn(expected_error_msg, str(cm.exception))

    def clean(self):
        super(FlowRuleException, self).clean()

        if (self.kind == flow_rule_kind.grading
                and self.expiration is not None):
            raise ValidationError(_("grading rules may not expire"))

        from course.validation import (
                ValidationError as ContentValidationError,
                validate_session_start_rule,
                validate_session_access_rule,
                validate_session_grading_rule,
                ValidationContext)
        from course.content import (get_course_repo,
                get_course_commit_sha,
                get_flow_desc)

        from relate.utils import dict_to_struct
        rule = dict_to_struct(self.rule)

        repo = get_course_repo(self.participation.course)
        commit_sha = get_course_commit_sha(
                self.participation.course, self.participation)
        ctx = ValidationContext(
                repo=repo,
                commit_sha=commit_sha)

        flow_desc = get_flow_desc(repo,
                self.participation.course,
                self.flow_id, commit_sha)

        tags = None
        if hasattr(flow_desc, "rules"):
            tags = getattr(flow_desc.rules, "tags", None)

        try:
            if self.kind == flow_rule_kind.start:
                validate_session_start_rule(ctx, unicode(self), rule, tags)
            elif self.kind == flow_rule_kind.access:
                validate_session_access_rule(ctx, unicode(self), rule, tags)
            elif self.kind == flow_rule_kind.grading:
                validate_session_grading_rule(ctx, unicode(self), rule, tags)
            else:
                # the rule refers to FlowRuleException rule
                raise ValidationError(_("invalid rule kind: ")+self.kind)

        except ContentValidationError as e:
            # the rule refers to FlowRuleException rule
            raise ValidationError(_("invalid existing_session_rules: ")+str(e))