Clean up the conftest for latest fixtures

Author: WilliamJamieson

tests/conftest.py

@@ -49,18 +49,23 @@ def _find_latest(uris):
 # Look directly at the latest schemas storage directory to infer latest schemas
 _LATEST_DIR = Path(__file__).parent.parent.absolute() / "latest"
 _LATEST_PATHS = MappingProxyType(
-    {latest_path: yaml.safe_load(latest_path.read_bytes()) for latest_path in _LATEST_DIR.glob("**/*.yaml")}
+    {
+        latest_path.relative_to(_LATEST_DIR): yaml.safe_load(latest_path.read_bytes())
+        for latest_path in _LATEST_DIR.glob("**/*.yaml")
+    }
 )
-_LATEST_TOP_LEVEL_PATHS = tuple(latest_path for latest_path in _LATEST_PATHS if latest_path.parent == _LATEST_DIR)
 _LATEST_URI_PATHS = MappingProxyType({schema["id"]: path for path, schema in _LATEST_PATHS.items()})
-_LATEST_URIS = tuple(_LATEST_URI_PATHS.keys())
-_LATEST_MANIFEST_URIS = tuple(uri for uri in _LATEST_URIS if "manifests" in uri)
+_LATEST_MANIFEST_URIS = MappingProxyType(
+    {schema["id"]: schema for schema in _LATEST_PATHS.values() if "manifests" in schema["id"]}
+)
+_LATEST_TOP_LEVEL_PATHS = tuple(latest_path for latest_path in _LATEST_PATHS if latest_path.parent == Path("."))
+
 _LATEST_MANIFEST_TAGS = MappingProxyType(
-    {uri: tuple(entry["tag_uri"] for entry in _CURRENT_RESOURCES[uri]["tags"]) for uri in _LATEST_MANIFEST_URIS}
+    {uri: tuple(entry["tag_uri"] for entry in schema["tags"]) for uri, schema in _LATEST_MANIFEST_URIS.items()}
 )
 _LATEST_DATAMODELS_URI = next(uri for uri in _LATEST_MANIFEST_URIS if "static" not in uri)
 _LATEST_STATIC_URI = next(uri for uri in _LATEST_MANIFEST_URIS if "static" in uri)
-_LATEST_DATAMODEL_URIS = tuple(uri["schema_uri"] for uri in _CURRENT_RESOURCES[_LATEST_DATAMODELS_URI]["tags"])
+_LATEST_DATAMODEL_URIS = tuple(uri["schema_uri"] for uri in _LATEST_MANIFEST_URIS[_LATEST_DATAMODELS_URI]["tags"])
 _LATEST_ARCHIVE_URIS = tuple(schema["id"] for schema in _LATEST_PATHS.values() if "archive_meta" in schema)
 
 
@@ -121,6 +126,22 @@ def latest_path(request):
     return request.param
 
 
+@pytest.fixture(scope="session")
+def latest_schema(latest_path):
+    """
+    Get the latest schema from a latest path.
+    """
+    return _LATEST_PATHS[latest_path]
+
+
+@pytest.fixture(scope="session")
+def latest_uri(latest_schema):
+    """
+    Get a latest resource URI
+    """
+    return latest_schema["id"]
+
+
 @pytest.fixture(scope="session", params=_LATEST_ARCHIVE_URIS)
 def latest_archive_uri(request):
     """
@@ -138,19 +159,27 @@ def latest_dir():
 
 
 @pytest.fixture(scope="session")
-def latest_reference_files_dir(latest_dir):
+def latest_datamodels_dir():
+    """
+    Get the path to the latest datamodels directory.
+    """
+    return Path(".")
+
+
+@pytest.fixture(scope="session")
+def latest_reference_files_dir(latest_datamodels_dir):
     """
     Get the path to the latest reference files directory.
     """
-    return latest_dir / "reference_files"
+    return latest_datamodels_dir / "reference_files"
 
 
 @pytest.fixture(scope="session")
-def latest_ccsp_dir(latest_dir):
+def latest_ccsp_dir(latest_datamodels_dir):
     """
     Get the path to the latest CCSP schemas directory.
     """
-    return latest_dir / "CCSP"
+    return latest_datamodels_dir / "CCSP"
 
 
 @pytest.fixture(scope="session", params=_LATEST_TOP_LEVEL_PATHS)
@@ -161,28 +190,12 @@ def latest_top_level_path(request):
     return request.param
 
 
-@pytest.fixture(scope="session")
-def latest_uri_paths():
-    """
-    Get the mapping of latest URIs to their paths.
-    """
-    return _LATEST_URI_PATHS
-
-
 @pytest.fixture(scope="session")
 def latest_uris():
     """
     Get the URIs of the latest schemas.
     """
-    return _LATEST_URIS
-
-
-@pytest.fixture(scope="session", params=_LATEST_URIS)
-def latest_uri(request):
-    """
-    Get a latest resource URI
-    """
-    return request.param
+    return _LATEST_URI_PATHS
 
 
 @pytest.fixture(scope="session")
@@ -235,11 +248,14 @@ def latest_static_tag_uri(request):
 
 
 @pytest.fixture(scope="session")
-def latest_schemas(latest_paths, latest_uris):
+def latest_schemas(latest_dir, latest_paths, latest_uris):
     """
     Get the text of the latest schemas.
     """
-    return {latest_uri: latest_path.read_text() for latest_uri, latest_path in zip(latest_uris, latest_paths, strict=True)}
+    return {
+        latest_uri: (latest_dir / latest_path).read_text()
+        for latest_uri, latest_path in zip(latest_uris, latest_paths, strict=True)
+    }
 
 
 @pytest.fixture(scope="session", params=tuple(entry["tag_uri"] for entry in _CURRENT_RESOURCES[_PREVIOUS_DATAMODELS_URI]["tags"]))
@@ -353,7 +369,7 @@ def schema_uris():
     return _SCHEMA_URIS
 
 
-@pytest.fixture(scope="session", params=tuple(uri for uri in _LATEST_URIS if uri in _SCHEMA_URIS))
+@pytest.fixture(scope="session", params=tuple(uri for uri in _LATEST_URI_PATHS if uri in _SCHEMA_URIS))
 def schema_uri(request):
     """
     Get a URI for a RAD schema from the ASDF resource manager.
@@ -366,15 +382,18 @@ def schema_uri(request):
     return request.param
 
 
-@pytest.fixture(scope="session", params=tuple(uri for uri in _LATEST_URIS if "/reference_files" in uri and uri in _SCHEMA_URIS))
+@pytest.fixture(
+    scope="session", params=tuple(uri for uri in _LATEST_URI_PATHS if "/reference_files" in uri and uri in _SCHEMA_URIS)
+)
 def ref_file_uri(request):
     """
-    Get a URI related to the RAD reference files from the ASDF resource manager.
+    Get a URI related to the RAD reference files from the ASDF r
+    esource manager.
     """
     return request.param
 
 
-@pytest.fixture(scope="session", params=tuple(uri for uri in _LATEST_URIS if "/CCSP" in uri and uri in _SCHEMA_URIS))
+@pytest.fixture(scope="session", params=tuple(uri for uri in _LATEST_URI_PATHS if "/CCSP" in uri and uri in _SCHEMA_URIS))
 def ccsp_uri(request):
     """
     Get a URI related to the RAD CCSP schemas.

tests/test_latest.py

@@ -10,7 +10,6 @@
 
 import asdf.treeutil
 import pytest
-import yaml
 
 from rad import resources
 from rad._parser import archive_schema, super_schema
@@ -28,16 +27,15 @@ def test_smoke_latest_paths(self, latest_paths):
         """
         assert latest_paths
 
-    def test_latest_filename(self, latest_path):
+    def test_latest_filename(self, latest_path, latest_uri):
         """
         Check that the file name of the schema matches the schema ID WITHOUT the version number suffix.
         """
         # Check that the file name does not contain a version number
-        assert len(str(latest_path).split("/latest/")[-1].split("-")) == 1
+        assert len(str(latest_path).split("-")) == 1
 
         # Check that the file name is consistent with the schema ID
-        uri = yaml.safe_load(latest_path.read_bytes())["id"]
-        uri = uri.split("/schemas/")[-1] if "/schemas/" in uri else uri.split("/manifests/")[-1]
+        uri = latest_uri.split("/schemas/")[-1] if "/schemas/" in latest_uri else latest_uri.split("/manifests/")[-1]
 
         # Find the filename without the version number suffix
         filename = str(latest_path.parent / latest_path.stem).split("/latest/")[-1].split("-")[0]
@@ -46,12 +44,12 @@ def test_latest_filename(self, latest_path):
         assert uri.split("-")[0] == filename.split("manifests/")[-1] if "manifests/" in filename else filename
 
     @pytest.mark.skipif(not IS_EDITABLE, reason="Symbolic links are resolved in non-editable installs")
-    def test_latest_symlink(self, latest_path):
+    def test_latest_symlink(self, latest_uri, latest_path, latest_dir):
         """
         Check that each "latest" file has a simlink in the `schemas` or `manifests` directory
         that points to that file which includes its version number (e.g. matches its ID).
         """
-        path_suffix = yaml.safe_load(latest_path.read_bytes())["id"].split("/roman/")[-1]
+        path_suffix = latest_uri.split("/roman/")[-1]
         path_prefix = importlib_resources.files(resources)
         symlink_path = path_prefix / f"{path_suffix}.yaml"
 
@@ -64,7 +62,9 @@ def test_latest_symlink(self, latest_path):
         assert not symlink_path.readlink().is_absolute(), f"Expected {symlink_path} to be a relative symlink, but it is not."
 
         # Check that the symlink points to the correct file
-        assert symlink_path.resolve() == latest_path, f"Expected {symlink_path} to point to {latest_path}, but it does not."
+        assert symlink_path.resolve() == latest_dir / latest_path, (
+            f"Expected {symlink_path} to point to {latest_dir / latest_path}, but it does not."
+        )
 
     def test_latest_datamodels_not_static(self, latest_datamodels_tag_uri, latest_static_tags):
         """
@@ -97,12 +97,12 @@ def test_latest_datamodels_only_reference_latest(self, latest_datamodels_tag_uri
             f"{tag_schema_map[latest_datamodels_tag_uri]} is not in latest uris."
         )
 
-    def test_latest_schemas(self, latest_schema_tags, latest_schemas, latest_uri):
+    def test_latest_schemas(self, latest_schema_tags, latest_schemas, latest_schema, latest_uri):
         """
         Check that the latest schemas are using the latest versions of the schema and tag uris.
         """
         # Sanity check that the latest_uri matches the id in the latest_schemas
-        assert yaml.safe_load(latest_schemas[latest_uri])["id"] == latest_uri
+        assert latest_schema["id"] == latest_uri
 
         # Substitute all matching patterns with the latest_uri and check equality
         # If any substitution changes the schema, then we have a URI mismatch
@@ -139,12 +139,12 @@ def test_no_internal_tags(self, latest_tagged_schema_uri, current_resources):
         )
 
     def test_tagged_schema_path(
-        self, latest_tagged_schema_uri, latest_dir, latest_reference_files_dir, latest_ccsp_dir, latest_uri_paths
+        self, latest_tagged_schema_uri, latest_datamodels_dir, latest_reference_files_dir, latest_ccsp_dir, latest_uris
     ):
         """
         Check that all datamodels are in the top-level directory and all reference files are in the reference_files directory.
         """
-        path = latest_uri_paths[latest_tagged_schema_uri]
+        path = latest_uris[latest_tagged_schema_uri]
         if "reference_file" in latest_tagged_schema_uri:
             assert path.parent == latest_reference_files_dir, (
                 f"{latest_tagged_schema_uri} is a reference file that is not in the reference_files directory."
@@ -159,7 +159,10 @@ def test_tagged_schema_path(
                 f"{latest_tagged_schema_uri} is not prefixed with '{ccsp_id_dir.lower()}_'"
             )
         else:
-            assert path.parent == latest_dir, f"{latest_tagged_schema_uri} is a datamodel that is not in the top-level directory."
+            print(path.parent)
+            assert path.parent == latest_datamodels_dir, (
+                f"{latest_tagged_schema_uri} is a datamodel that is not in the top-level directory."
+            )
 
     def test_top_level_schema(self, latest_top_level_path, metaschema_uri, latest_paths):
         """