Fix #8544: Parse the correct schema version from manifest

.changes/unreleased/Fixes-20230906-162427.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Parse the correct schema version from manifest
+time: 2023-09-06T16:24:27.849069+01:00
+custom:
+  Author: aranke
+  Issue: "8544"

core/dbt/contracts/graph/manifest.py

@@ -1528,7 +1528,15 @@
     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])
+
+    # schema_version is in this format: https://schemas.getdbt.com/dbt/manifest/v10.json
+    # What the code below is doing:
+    # 1. Split on "/" – v10.json
+    # 2. Split on "." – v10
+    # 3. Skip first character – 10
+    # 4. Convert to int
+    # TODO: If this gets more complicated, turn into a regex
+    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