Update test

Author: Rafnuss

.github/workflows/validate-profile-and-schemas.yml

@@ -2,9 +2,7 @@ name: Validate profile and schemas
 
 on:
   push:
-    branches: [main]
   pull_request:
-    branches: [main]
 
 jobs:
   validate-profile-and-schemas:

README.md

@@ -35,3 +35,12 @@ Create a production-style build:
 ```bash
 JEKYLL_ENV=production bundle exec jekyll build
 ```
+
+Run validation tests:
+
+```bash
+python3 -m venv .venv
+source .venv/bin/activate
+python -m pip install -r tests/requirements.txt
+python tests/validate_profile_and_schemas.py
+```

tests/helpers.py

@@ -1,46 +1,2 @@
-attrs==23.1.0
-black==21.9b0
-certifi==2021.10.8
-chardet==4.0.0
-charset-normalizer==2.0.7
-click==8.0.3
-colorama==0.4.4
-decorator==5.1.0
+# Minimal dependencies for tests/validate_profile_and_schemas.py
 frictionless>=5.13.1
-humanize==4.6.0
-idna==3.3
-isodate==0.6.0
-Jinja2==3.1.2
-jsonschema>=4.20
-markdown-it-py==2.2.0
-marko==1.1.0
-MarkupSafe==2.1.2
-mdurl==0.1.2
-mypy==0.910
-mypy-extensions==0.4.3
-pathspec==0.9.0
-petl==1.7.4
-platformdirs==2.4.0
-pydantic>=2.0
-Pygments==2.15.1
-pyrsistent==0.18.0
-python-dateutil==2.8.2
-python-slugify==5.0.2
-PyYAML==6.0
-regex==2021.10.23
-reportbro-simpleeval==0.9.11
-requests==2.26.0
-rfc3986==1.5.0
-rich==13.3.5
-shellingham==1.4.0
-simpleeval==0.9.11
-six==1.16.0
-stringcase==1.2.0
-tabulate==0.9.0
-text-unidecode==1.3
-toml==0.10.2
-tomli==1.2.1
-typer>=0.12
-typing_extensions==4.6.1
-urllib3==1.26.7
-validators==0.18.2

tests/requirements.txt

@@ -3,19 +3,40 @@
 import sys
 import json
 from pathlib import Path
-from typing import List
+from typing import Dict, List, Optional
 from frictionless import Schema
-from helpers import PROFILE_PATH, TABLE_SCHEMA_PATHS
 
+THIS_SCRIPT_PATH = Path(__file__).parent
+REPOSITORY_ROOT_PATH = THIS_SCRIPT_PATH / ".."
+PROFILE_PATH = REPOSITORY_ROOT_PATH / "geolocator-dp-profile.json"
+TABLE_SCHEMA_PATHS = [
+    REPOSITORY_ROOT_PATH / "observations-table-schema.json",
+    REPOSITORY_ROOT_PATH / "tags-table-schema.json",
+    REPOSITORY_ROOT_PATH / "measurements-table-schema.json",
+    REPOSITORY_ROOT_PATH / "staps-table-schema.json",
+    REPOSITORY_ROOT_PATH / "twilights-table-schema.json",
+    REPOSITORY_ROOT_PATH / "paths-table-schema.json",
+    REPOSITORY_ROOT_PATH / "edges-table-schema.json",
+    REPOSITORY_ROOT_PATH / "pressurepaths-table-schema.json",
+]
 
-def validate_json(filepath: Path) -> bool:
+EXPECTED_SCHEMA_RESOURCES = {
+    "tags",
+    "observations",
+    "measurements",
+    "staps",
+    "twilights",
+    "paths",
+    "edges",
+    "pressurepaths",
+}
+
+def load_json(filepath: Path) -> Optional[dict]:
     with open(filepath) as file:
         try:
-            json.load(file)
+            return json.load(file)
         except json.decoder.JSONDecodeError:
-            return False
-        else:
-            return True
+            return None
 
 
 def validate_schema(file_path: Path) -> bool:
@@ -32,12 +53,124 @@ def get_schema_metadata_error_messages(file_path: Path) -> List[str]:
     return [err.message for err in report.errors]
 
 
+def _listify_fields(fields) -> List[str]:
+    if isinstance(fields, str):
+        return [fields]
+    if isinstance(fields, list):
+        return [field for field in fields if isinstance(field, str)]
+    return []
+
+
+def _get_schema_fields(descriptor: dict) -> set:
+    fields = descriptor.get("fields", [])
+    return {
+        field.get("name")
+        for field in fields
+        if isinstance(field, dict) and isinstance(field.get("name"), str)
+    }
+
+
+def check_schema_coherence(schema_descriptors: Dict[Path, dict]) -> bool:
+    encountered_errors = False
+    resource_to_paths: Dict[str, List[Path]] = {}
+
+    # 1) filename <-> schema name consistency
+    for schema_path, descriptor in schema_descriptors.items():
+        schema_name = descriptor.get("name")
+        expected_name = schema_path.name.replace("-table-schema.json", "")
+
+        if not isinstance(schema_name, str):
+            print(f"✕ {schema_path.name}: missing or non-string `name`")
+            encountered_errors = True
+            continue
+
+        resource_to_paths.setdefault(schema_name, []).append(schema_path)
+
+        if schema_name != expected_name:
+            print(
+                f"✕ {schema_path.name}: schema `name` is `{schema_name}` but expected `{expected_name}`"
+            )
+            encountered_errors = True
+
+    # 2) expected resources exist exactly once
+    observed_resources = set(resource_to_paths.keys())
+    missing_resources = sorted(EXPECTED_SCHEMA_RESOURCES - observed_resources)
+    unexpected_resources = sorted(observed_resources - EXPECTED_SCHEMA_RESOURCES)
+
+    for resource in missing_resources:
+        print(f"✕ schema resources: missing expected resource `{resource}`")
+        encountered_errors = True
+
+    for resource in unexpected_resources:
+        print(f"✕ schema resources: unexpected resource `{resource}`")
+        encountered_errors = True
+
+    for resource, paths in sorted(resource_to_paths.items()):
+        if len(paths) > 1:
+            path_list = ", ".join(path.name for path in paths)
+            print(
+                f"✕ schema resources: resource `{resource}` is defined more than once ({path_list})"
+            )
+            encountered_errors = True
+
+    # 3) foreign keys reference existing resources and target fields
+    fields_by_resource: Dict[str, set] = {}
+    for resource, paths in resource_to_paths.items():
+        if len(paths) == 1:
+            fields_by_resource[resource] = _get_schema_fields(schema_descriptors[paths[0]])
+
+    for schema_path, descriptor in schema_descriptors.items():
+        foreign_keys = descriptor.get("foreignKeys", [])
+        if not isinstance(foreign_keys, list):
+            continue
+
+        for foreign_key in foreign_keys:
+            if not isinstance(foreign_key, dict):
+                continue
+
+            reference = foreign_key.get("reference", {})
+            if not isinstance(reference, dict):
+                print(f"✕ {schema_path.name}: malformed foreign key reference")
+                encountered_errors = True
+                continue
+
+            target_resource = reference.get("resource")
+            target_fields = _listify_fields(reference.get("fields"))
+
+            if not isinstance(target_resource, str):
+                print(f"✕ {schema_path.name}: foreign key reference missing `resource`")
+                encountered_errors = True
+                continue
+
+            if target_resource not in resource_to_paths:
+                print(
+                    f"✕ {schema_path.name}: foreign key references unknown resource `{target_resource}`"
+                )
+                encountered_errors = True
+                continue
+
+            target_resource_fields = fields_by_resource.get(target_resource, set())
+            missing_target_fields = [
+                field for field in target_fields if field not in target_resource_fields
+            ]
+
+            if missing_target_fields:
+                fields_str = ", ".join(missing_target_fields)
+                print(
+                    f"✕ {schema_path.name}: foreign key references missing field(s) in `{target_resource}`: {fields_str}"
+                )
+                encountered_errors = True
+
+    return not encountered_errors
+
+
 if __name__ == "__main__":
     encountered_errors = False
+    schema_descriptors: Dict[Path, dict] = {}
 
     print(PROFILE_PATH.name)
-    result = validate_json(PROFILE_PATH)
-    if result is True:
+    profile_json = load_json(PROFILE_PATH)
+    if profile_json is not None:
         print("✔︎ valid JSON")
     else:
         print("✕ valid JSON")
@@ -46,7 +179,9 @@ def get_schema_metadata_error_messages(file_path: Path) -> List[str]:
     for table_schema in TABLE_SCHEMA_PATHS:
         print(f"\n{table_schema.name}")
 
-        if validate_json(table_schema):
+        schema_json = load_json(table_schema)
+        if schema_json is not None:
+            schema_descriptors[table_schema] = schema_json
             print("✔︎ valid JSON")
             if validate_schema(table_schema):
                 print("✔︎ valid Table Schema")
@@ -59,6 +194,12 @@ def get_schema_metadata_error_messages(file_path: Path) -> List[str]:
             print("✕ valid JSON")
             encountered_errors = True
 
+    print("\nSchema coherence")
+    if check_schema_coherence(schema_descriptors):
+        print("✔︎ schema coherence checks passed")
+    else:
+        encountered_errors = True
+
     if encountered_errors:
         print("Errors were encountered")
         sys.exit(1)