validation.py

def check_attributes_yml(vctx, repo, path, tree):
    """
    This function reads the .attributes.yml file and checks
    that each item for each header is a string

    att_roles : list
         list of acceptable access types

    Example::

        # this validates
        unenrolled:
            - test1.pdf
        student:
            - test2.pdf

        # this does not validate
        unenrolled:
            - test1.pdf
        student:
            - test2.pdf
            - 42
    """
    try:
        dummy, attr_blob_sha = tree[b".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"

        att_roles = ["public", "in_exam", "student", "ta",
                     "unenrolled", "instructor"]
        validate_struct(vctx, loc, att_yml,
                        required_attrs=[],
                        allowed_attrs=[(role, list) for role in att_roles])

        if hasattr(att_yml, "public"):
            vctx.add_warning(loc,
                    _("Access class 'public' is deprecated. Use 'unenrolled' "
                        "instead."))

        if hasattr(att_yml, "public") and hasattr(att_yml, "unenrolled"):
            raise ValidationError(
                _("%s: access classes 'public' and 'unenrolled' may not "
                    "exist simultaneously.")
                % (loc))

        for access_kind in att_roles:
            if hasattr(att_yml, access_kind):
                for i, l in enumerate(getattr(att_yml, access_kind)):
                    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):
            dummy, blob_sha = tree[entry.path]
            subtree = repo[blob_sha]
            check_attributes_yml(vctx, repo,
                                 path+"/"+entry.path.decode("utf-8"), subtree)


# {{{ check whether flow grade identifiers were changed in sketchy ways

def check_grade_identifier_link(
        vctx, location, course, flow_id, flow_grade_identifier):

    from course.models import GradingOpportunity
    for bad_gopp in (
            GradingOpportunity.objects
            .filter(
                course=course,
                identifier=flow_grade_identifier)
            .exclude(flow_id=flow_id)):
        # 0 or 1 trips through this loop because of uniqueness

        raise ValidationError(
                _(
                    "{location}: existing grading opportunity with identifier "
                    "'{grade_identifier}' refers to flow '{other_flow_id}', however "
                    "flow code in this flow ('{new_flow_id}') specifies the same "
                    "grade identifier. "
                    "(Have you renamed the flow? If so, edit the grading "
                    "opportunity to match.)")
                .format(
                    location=location,
                    grade_identifier=flow_grade_identifier,
                    other_flow_id=bad_gopp.flow_id,
                    new_flow_id=flow_id,
                    new_grade_identifier=flow_grade_identifier))

# }}}


# {{{ check whether page types were changed

def check_for_page_type_changes(vctx, location, course, flow_id, flow_desc):
    from course.content import normalize_flow_desc
    n_flow_desc = normalize_flow_desc(flow_desc)

    from course.models import FlowPageData
    for grp in n_flow_desc.groups:
        for page_desc in grp.pages:
            fpd_with_mismatched_page_types = list(
                    FlowPageData.objects
                    .filter(
                        flow_session__course=course,
                        flow_session__flow_id=flow_id,
                        group_id=grp.id,
                        page_id=page_desc.id)
                    .exclude(page_type=None)
                    .exclude(page_type=page_desc.type)
                    [0:1])

            if fpd_with_mismatched_page_types:
                mismatched_fpd, = fpd_with_mismatched_page_types
                raise ValidationError(
                        _("%(loc)s, group '%(group)s', page '%(page)s': "
                            "page type ('%(type_new)s') differs from "
                            "type used in database ('%(type_old)s')")
                        % {"loc": location, "group": grp.id,
                            "page": page_desc.id,
                            "type_new": page_desc.type,
                            "type_old": mismatched_fpd.page_type})

# }}}


def validate_course_content(repo, course_file, events_file,
        validate_sha, course=None):
    vctx = ValidationContext(
            repo=repo,
            commit_sha=validate_sha,
            course=course)

    course_desc = get_yaml_from_repo_safely(repo, course_file,
            commit_sha=validate_sha)

    validate_staticpage_desc(vctx, course_file, course_desc)

    try:
        from course.content import get_yaml_from_repo
        events_desc = get_yaml_from_repo(repo, events_file,
                commit_sha=validate_sha, cached=False)
    except ObjectDoesNotExist:
        if events_file != "events.yml":
            vctx.add_warning(
                    _("Events file"),
                    _("Your course repository does not have an events "
                        "file named '%s'.")
                    % events_file)
        else:
            # That's OK--no calendar info.
            pass
    else:
        validate_calendar_desc_struct(vctx, events_file, events_desc)

    check_attributes_yml(
            vctx, repo, "", get_repo_blob(repo, "", validate_sha))

    try:
        flows_tree = get_repo_blob(repo, "media", validate_sha)
    except ObjectDoesNotExist:
        # That's great--no media directory.
        pass
    else:
        vctx.add_warning(
                'media/', _(
                    "Your course repository has a 'media/' directory. "
                    "Linking to media files using 'media:' is discouraged. "
                    "Use the 'repo:' and 'repocur:' linkng schemes instead."))

    # {{{ flows

    try:
        flows_tree = get_repo_blob(repo, "flows", validate_sha)
    except ObjectDoesNotExist:
        # That's OK--no flows yet.
        pass
    else:
        used_grade_identifiers = set()

        for entry in flows_tree.items():
            entry_path = entry.path.decode("utf-8")
            if not entry_path.endswith(".yml"):
                continue

            from course.constants import FLOW_ID_REGEX
            flow_id = entry_path[:-4]
            match = re.match("^"+FLOW_ID_REGEX+"$", flow_id)
            if match is None:
                raise ValidationError(
                        string_concat("%s: ",
                            _("invalid flow name. "
                                "Flow names may only contain (roman) "
                                "letters, numbers, "
                                "dashes and underscores."))
                        % entry_path)

            location = "flows/%s" % entry_path
            flow_desc = get_yaml_from_repo_safely(repo, location,
                    commit_sha=validate_sha)

            validate_flow_desc(vctx, location, flow_desc)

            # {{{ check grade_identifier

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

            if (
                    flow_grade_identifier is not None
                    and
                    set([flow_grade_identifier]) & used_grade_identifiers):
                raise ValidationError(
                        string_concat("%s: ",
                                      _("flow uses the same grade_identifier "
                                        "as another flow"))
                        % location)

            used_grade_identifiers.add(flow_grade_identifier)

            if (course is not None
                    and flow_grade_identifier is not None):
                check_grade_identifier_link(
                        vctx, location, course, flow_id, flow_grade_identifier)

            # }}}

            if course is not None:
                check_for_page_type_changes(
                        vctx, location, course, flow_id, flow_desc)

    # }}}

    # {{{ static pages

    try:
        pages_tree = get_repo_blob(repo, "staticpages", validate_sha)
    except ObjectDoesNotExist:
        # That's OK--no flows yet.
        pass
    else:
        for entry in pages_tree.items():
            entry_path = entry.path.decode("utf-8")
            if not entry_path.endswith(".yml"):
                continue

            from course.constants import STATICPAGE_PATH_REGEX
            page_name = entry_path[:-4]
            match = re.match("^"+STATICPAGE_PATH_REGEX+"$", page_name)
            if match is None:
                raise ValidationError(
                        string_concat("%s: ",
                            _(
                                "invalid page name. "
                                "Page names may only contain "
                                "alphanumeric characters (any language) "
                                "and hyphens."
                                ))
                        % entry_path)

        location = "staticpages/%s" % entry_path
        page_desc = get_yaml_from_repo_safely(repo, location,
                commit_sha=validate_sha)

        validate_staticpage_desc(vctx, location, page_desc)

    # }}}

    return vctx.warnings


# {{{ validation script support

class FileSystemFakeRepo(object):
    def __init__(self, root):
        self.root = root
        assert isinstance(self.root, six.binary_type)

    def controldir(self):
        return self.root

    def __getitem__(self, sha):
        return sha

    def __str__(self):
        return "<FAKEREPO:%s>" % self.root

    def decode(self):
        return self

    @property
    def tree(self):
        return FileSystemFakeRepoTree(self.root)


class FileSystemFakeRepoTreeEntry(object):
    def __init__(self, path, mode):
        self.path = path
        self.mode = mode


class FileSystemFakeRepoTree(object):
    def __init__(self, root):
        self.root = root
        assert isinstance(self.root, six.binary_type)

    def __getitem__(self, name):
        if not name:
            raise KeyError("<empty filename>")

        from os.path import join, isdir, exists
        name = join(self.root, name)

        if not exists(name):
            raise KeyError(name)

        # returns mode, "sha"
        if isdir(name):
            return None, FileSystemFakeRepoTree(name)
        else:
            return None, FileSystemFakeRepoFile(name)

    def items(self):
        import os
        return [
                FileSystemFakeRepoTreeEntry(
                    path=n,
                    mode=os.stat(os.path.join(self.root, n)).st_mode)
                for n in os.listdir(self.root)]


class FileSystemFakeRepoFile(object):
    def __init__(self, name):
        self.name = name

    @property
    def data(self):
        with open(self.name, "rb") as inf:
            return inf.read()


def validate_course_on_filesystem(
        root, course_file, events_file):
    fake_repo = FileSystemFakeRepo(root.encode("utf-8"))
    warnings = validate_course_content(
            fake_repo,
            course_file, events_file,
            validate_sha=fake_repo, course=None)

    if warnings:
        print(_("WARNINGS: "))
        for w in warnings:
            print("***", w.location, w.text)

# }}}

# vim: foldmethod=marker