Fix #8551: Parse the correct schema version from manifest

core/dbt/contracts/graph/manifest.py

@@ -1528,7 +1528,8 @@ def get_manifest_schema_version(dct: dict) -> int:
     schema_version = dct.get("metadata", {}).get("dbt_schema_version", None)
     if not schema_version:
         raise ValueError("Manifest doesn't have schema version")
-    return int(schema_version.split(".")[-2][-1])
+
+    return int(schema_version.split("/")[-1].split(".")[0][1:])
 
 
 def _check_duplicates(value: BaseNode, src: Mapping[str, BaseNode]):

tests/functional/artifacts/test_previous_version_state.py

@@ -1,9 +1,12 @@
-import pytest
+import json
 import os
 import shutil
-from dbt.tests.util import run_dbt, get_manifest
+
+import pytest
+
+from dbt.contracts.graph.manifest import WritableManifest, get_manifest_schema_version
 from dbt.exceptions import IncompatibleSchemaError
-from dbt.contracts.graph.manifest import WritableManifest
+from dbt.tests.util import run_dbt, get_manifest
 
 # This project must have one of each kind of node type, plus disabled versions, for
 # test coverage to be complete.
@@ -352,3 +355,13 @@ def test_nonbackwards_compatible_versions(self, project):
         # schema versions 1, 2, 3 are all not forward compatible
         for schema_version in range(1, 4):
             self.compare_previous_state(project, schema_version, False, 0)
+
+    def test_get_manifest_schema_version(self, project):
+        for schema_version in range(1, self.CURRENT_EXPECTED_MANIFEST_VERSION):
+            manifest_path = os.path.join(
+                project.test_data_dir, f"state/v{schema_version}/manifest.json"
+            )
+            manifest = json.load(open(manifest_path))
+
+            manifest_version = get_manifest_schema_version(manifest)
+            assert manifest_version == schema_version