import json import re import sys from collections import defaultdict from typing import Any, Callable, Dict, List, Optional, Set, Tuple import fastjsonschema import yaml MODELS_YML_PATH = "../../docs/models.yml" DEFAULT_FILES = [ "../../docker/initial-data.json", "../../docs/example-data.json", ] SCHEMA = fastjsonschema.compile( { "$schema": "http://json-schema.org/draft-07/schema#", "title": "Schema for initial and example data.", "type": "object", "patternProperties": { "^[a-z_]+$": { "type": "array", "items": { "type": "object", "properties": {"id": {"type": "number"}}, "required": ["id"], }, } }, "additionalProperties": False, } ) class CheckException(Exception): pass def check_string(value: Any) -> bool: return value is None or isinstance(value, str) color_regex = re.compile("^#[0-9a-f]{6}$") def check_color(value: Any) -> bool: return value is None or bool(isinstance(value, str) and color_regex.match(value)) def check_number(value: Any) -> bool: return value is None or isinstance(value, int) def check_float(value: Any) -> bool: return value is None or isinstance(value, int) or isinstance(value, float) def check_boolean(value: Any) -> bool: return value is None or value is False or value is True def check_string_list(value: Any) -> bool: return check_x_list(value, check_string) def check_number_list(value: Any) -> bool: return check_x_list(value, check_number) def check_x_list(value: Any, fn: Callable) -> bool: if value is None: return True if not isinstance(value, list): return False return all([fn(sv) for sv in value]) def check_decimal(value: Any) -> bool: if value is None: return True if isinstance(value, str): pattern = r"^-?(\d|[1-9]\d+)\.\d{6}$" if re.match(pattern, value): return True return False def check_json(value: Any, root: bool = True) -> bool: if value is None: return True if not root and (isinstance(value, int) or isinstance(value, str)): return True if isinstance(value, list): return all(check_json(x, root=False) for x in value) if isinstance(value, dict): return all(check_json(x, root=False) for x in value.values()) return False checker_mapping = { "string": check_string, "HTMLStrict": check_string, "HTMLPermissive": check_string, "generic-relation": check_string, "number": check_number, "timestamp": check_number, "relation": check_number, "float": check_float, "boolean": check_boolean, "string[]": check_string_list, "generic-relation-list": check_string_list, "number[]": check_number_list, "relation-list": check_number_list, "decimal(6)": check_decimal, "color": check_color, "JSON": check_json, } class Checker: def __init__(self, data: Dict[str, List[Any]], is_import: bool = False) -> None: self.data = data with open(MODELS_YML_PATH, "rb") as x: models_yml = x.read() models_yml = models_yml.replace(" yes:".encode(), ' "yes":'.encode()) models_yml = models_yml.replace(" no:".encode(), ' "no":'.encode()) self.models = yaml.safe_load(models_yml) if is_import: self.modify_models_for_import() self.errors: List[str] = [] self.template_prefixes: Dict[ str, Dict[str, Tuple[str, int, int]] ] = defaultdict(dict) self.generate_template_prefixes() def modify_models_for_import(self) -> None: collection_allowlist = ( "user", "meeting", "group", "personal_note", "tag", "agenda_item", "list_of_speakers", "speaker", "topic", "motion", "motion_submitter", "motion_comment", "motion_comment_section", "motion_category", "motion_block", "motion_change_recommendation", "motion_state", "motion_workflow", "motion_statute_paragraph", "poll", "option", "vote", "assignment", "assignment_candidate", "mediafile", "projector", "projection", "projector_message", "projector_countdown", "chat_group", ) for collection in list(self.models.keys()): if collection not in collection_allowlist: del self.models[collection] self.models["mediafile"]["blob"] = "string" def generate_template_prefixes(self) -> None: for collection in self.models.keys(): for field in self.models[collection]: if not self.is_template_field(field): continue parts = field.split("$") prefix = parts[0] suffix = parts[1] if prefix in self.template_prefixes[collection]: raise ValueError( f"the template prefix {prefix} is not unique within {collection}" ) self.template_prefixes[collection][prefix] = ( field, len(prefix), len(suffix), ) def is_template_field(self, field: str) -> bool: return "$_" in field or field.endswith("$") def is_structured_field(self, field: str) -> bool: return "$" in field and not self.is_template_field(field) def is_normal_field(self, field: str) -> bool: return "$" not in field def make_structured(self, field: str, replacement: Any) -> str: if type(replacement) not in (str, int): raise CheckException( f"Invalid type {type(replacement)} for the replacement of field {field}" ) parts = field.split("$") return parts[0] + "$" + str(replacement) + parts[1] def to_template_field( self, collection: str, structured_field: str ) -> Tuple[str, str]: """Returns template_field, replacement""" parts = structured_field.split("$") descriptor = self.template_prefixes[collection].get(parts[0]) if not descriptor: raise CheckException( f"Unknown template field for prefix {parts[0]} in collection {collection}" ) return ( descriptor[0], structured_field[descriptor[1] + 1 : len(structured_field) - descriptor[2]], ) def run_check(self) -> None: self.check_json() self.check_collections() for collection, models in self.data.items(): for model in models: self.check_model(collection, model) if self.errors: errors = [f"\t{error}" for error in self.errors] raise CheckException("\n".join(errors)) def check_json(self) -> None: try: SCHEMA(self.data) except fastjsonschema.exceptions.JsonSchemaException as e: raise CheckException(f"JSON does not match schema: {str(e)}") def check_collections(self) -> None: c1 = set(self.data.keys()) c2 = set(self.models.keys()) if c1 != c2: err = "Collections in JSON file do not match with models.yml." if c2 - c1: err += f" Missing collections: {', '.join(c2-c1)}." if c1 - c2: err += f" Invalid collections: {', '.join(c1-c2)}." raise CheckException(err) def check_model(self, collection: str, model: Dict[str, Any]) -> None: errors = self.check_normal_fields(model, collection) if not errors: errors = self.check_template_fields(model, collection) if not errors: self.check_types(model, collection) self.check_relations(model, collection) def check_normal_fields(self, model: Dict[str, Any], collection: str) -> bool: model_fields = set( x for x in model.keys() if self.is_normal_field(x) or self.is_template_field(x) ) collection_fields = set( x for x in self.models[collection].keys() if self.is_normal_field(x) or self.is_template_field(x) ) errors = False if collection_fields - model_fields: error = f"{collection}/{model['id']}: Missing fields {', '.join(collection_fields - model_fields)}" self.errors.append(error) errors = True if model_fields - collection_fields: error = f"{collection}/{model['id']}: Invalid fields {', '.join(model_fields - collection_fields)}" self.errors.append(error) errors = True return errors def check_template_fields(self, model: Dict[str, Any], collection: str) -> bool: """ Only checks that for each replacement a structured field exists and not too many structured fields. Does not check the content. Returns True on errors. """ errors = False for template_field in self.models[collection].keys(): if not self.is_template_field(template_field): continue field_error = False replacements = model[template_field] if not isinstance(replacements, list): self.errors.append( f"{collection}/{model['id']}/{template_field}: Replacements for the template field must be a list" ) field_error = True continue for replacement in replacements: if not isinstance(replacement, str): self.errors.append( f"{collection}/{model['id']}/{template_field}: Each replacement for the template field must be a string" ) field_error = True if field_error: error = True continue replacement_collection = None field_description = self.models[collection][template_field] if isinstance(field_description, dict): replacement_collection = field_description.get("replacement_collection") for replacement in replacements: structured_field = self.make_structured(template_field, replacement) if structured_field not in model: self.errors.append( f"{collection}/{model['id']}/{template_field}: Missing {structured_field} since it is given as a replacement" ) errors = True if replacement_collection: try: as_id = int(replacement) except (TypeError, ValueError): self.errors.append( f"{collection}/{model['id']}/{template_field}: Replacement {replacement} is not an integer" ) if not self.find_model(replacement_collection, as_id): self.errors.append( f"{collection}/{model['id']}/{template_field}: Replacement {replacement} does not exist as a model of collection {replacement_collection}" ) for field in model.keys(): if self.is_structured_field(field): try: _template_field, _replacement = self.to_template_field( collection, field ) if ( template_field == _template_field and _replacement not in model[template_field] ): self.errors.append( f"{collection}/{model['id']}/{field}: Invalid structured field. Missing replacement {_replacement} in {template_field}" ) errors = True except CheckException as e: self.errors.append( f"{collection}/{model['id']}/{field} error: " + str(e) ) errors = True return errors def check_types(self, model: Dict[str, Any], collection: str) -> None: for field in model.keys(): if self.is_template_field(field): continue field_type = self.get_type_from_collection(field, collection) enum = self.get_enum_from_collection_field(field, collection) checker = checker_mapping.get(field_type) if checker is None: raise NotImplementedError( f"TODO implement check for field type {field_type}" ) if not checker(model[field]): error = f"{collection}/{model['id']}/{field}: Type error: Type is not {field_type}" self.errors.append(error) if enum and model[field] not in enum: error = f"{collection}/{model['id']}/{field}: Value error: Value {model[field]} is not a valid enum value" self.errors.append(error) def get_type_from_collection(self, field: str, collection: str) -> str: if self.is_structured_field(field): field, _ = self.to_template_field(collection, field) field_description = self.models[collection][field] if isinstance(field_description, dict): if field_description["type"] == "template": if isinstance(field_description["fields"], dict): return field_description["fields"]["type"] return field_description["fields"] return field_description["type"] return field_description field_description = self.get_field_description(field, collection) if field_description: return field_description["type"] return self.models[collection][field] def get_enum_from_collection_field( self, field: str, collection: str ) -> Optional[Set[str]]: if self.is_structured_field(field): field, _ = self.to_template_field(collection, field) field_description = self.models[collection][field] if isinstance(field_description, dict): if field_description["type"] == "template": if isinstance(field_description["fields"], dict): field_description = field_description["fields"] else: return None if "enum" in field_description: enum = set(field_description["enum"]) if not field_description.get("required", False): enum.add(None) return enum return None def check_relations(self, model: Dict[str, Any], collection: str) -> None: for field in model.keys(): try: self.check_relation(model, collection, field) except CheckException as e: self.errors.append( f"{collection}/{model['id']}/{field} error: " + str(e) ) def check_relation( self, model: Dict[str, Any], collection: str, field: str ) -> None: if self.is_template_field(field): return field_type = self.get_type_from_collection(field, collection) basemsg = f"{collection}/{model['id']}/{field}: Relation Error: " replacement = None if self.is_structured_field(field): _, replacement = self.to_template_field(collection, field) if field_type == "relation": foreign_id = model[field] if foreign_id is None: return foreign_collection, foreign_field = self.get_to(field, collection) self.check_reverse_relation( collection, model["id"], model, foreign_collection, foreign_id, foreign_field, basemsg, replacement, ) elif field_type == "relation-list": foreign_ids = model[field] if foreign_ids is None: return foreign_collection, foreign_field = self.get_to(field, collection) for foreign_id in foreign_ids: self.check_reverse_relation( collection, model["id"], model, foreign_collection, foreign_id, foreign_field, basemsg, replacement, ) elif field_type == "generic-relation" and model[field] is not None: foreign_collection, foreign_id = self.split_fqid(model[field]) foreign_field = self.get_to_generic_case( collection, field, foreign_collection ) self.check_reverse_relation( collection, model["id"], model, foreign_collection, foreign_id, foreign_field, basemsg, replacement, ) elif field_type == "generic-relation-list" and model[field] is not None: for fqid in model[field]: foreign_collection, foreign_id = self.split_fqid(fqid) foreign_field = self.get_to_generic_case( collection, field, foreign_collection ) self.check_reverse_relation( collection, model["id"], model, foreign_collection, foreign_id, foreign_field, basemsg, replacement, ) def get_to(self, field: str, collection: str) -> Tuple[str, str]: if self.is_structured_field(field): field, _ = self.to_template_field(collection, field) field_description = self.models[collection][field] if field_description["type"] == "template": to = field_description["fields"]["to"] else: to = field_description["to"] return to.split("/") def find_model(self, collection: str, id: int) -> Optional[Dict[str, Any]]: c = self.data.get(collection, []) for model in c: if model["id"] == id: return model return None def check_reverse_relation( self, collection: str, id: int, model: Dict[str, Any], foreign_collection: str, foreign_id: int, foreign_field: str, basemsg: str, replacement: Optional[str], ) -> None: actual_foreign_field = foreign_field if self.is_template_field(foreign_field): if replacement: actual_foreign_field = self.make_structured(foreign_field, replacement) else: replacement_collection = self.models[foreign_collection][foreign_field][ "replacement_collection" ] replacement = model.get(f"{replacement_collection}_id") if not replacement: self.errors.append( f"{basemsg} points to {foreign_collection}/{foreign_id}/{foreign_field}," f" but there is no replacement for {replacement_collection}" ) actual_foreign_field = self.make_structured(foreign_field, replacement) foreign_model = self.find_model(foreign_collection, foreign_id) foreign_value = ( foreign_model.get(actual_foreign_field) if foreign_model is not None else None ) foreign_field_type = self.get_type_from_collection( foreign_field, foreign_collection ) fqid = f"{collection}/{id}" error = False if foreign_field_type == "relation": error = foreign_value != id elif foreign_field_type == "relation-list": error = not foreign_value or id not in foreign_value elif foreign_field_type == "generic-relation": error = foreign_value != fqid elif foreign_field_type == "generic-relation-list": error = not foreign_value or fqid not in foreign_value else: raise NotImplementedError() if error: self.errors.append( f"{basemsg} points to {foreign_collection}/{foreign_id}/{actual_foreign_field}," " but the reverse relation for is corrupt" ) def split_fqid(self, fqid: str) -> Tuple[str, int]: try: collection, _id = fqid.split("/") id = int(_id) if collection not in self.models.keys(): raise CheckException(f"Fqid {fqid} has an invalid collection") return collection, id except (ValueError, AttributeError): raise CheckException(f"Fqid {fqid} is malformed") def split_collectionfield(self, collectionfield: str) -> Tuple[str, str]: collection, field = collectionfield.split("/") if collection not in self.models.keys(): raise CheckException( f"Collectionfield {collectionfield} has an invalid collection" ) if ( field not in self.models[collection] ): # Note: this has to be adopted when supporting template fields raise CheckException( f"Collectionfield {collectionfield} has an invalid field" ) return collection, field def get_to_generic_case( self, collection: str, field: str, foreign_collection: str ) -> str: """Returns all reverse relations as collectionfields""" to = self.models[collection][field]["to"] if isinstance(to, dict): if foreign_collection not in to["collections"]: raise CheckException( f"The collection {foreign_collection} is not supported " "as a reverse relation in {collection}/{field}" ) return to["field"] for cf in to: c, f = self.split_collectionfield(cf) if c == foreign_collection: return f raise CheckException( f"The collection {foreign_collection} is not supported as a reverse relation in {collection}/{field}" ) def main() -> int: files = sys.argv[1:] if not files: files = DEFAULT_FILES is_import = "--import" in files if is_import: files = [x for x in files if x != "--import"] failed = False for f in files: with open(f) as data: try: Checker(json.load(data), is_import=is_import).run_check() except CheckException as e: print(f"Check for {f} failed:\n", e) failed = True else: print(f"Check for {f} successful.") return 1 if failed else 0 if __name__ == "__main__": sys.exit(main())