Repository: SFDO-Tooling/CumulusCI
Pull request #3292 (2022-07-18 17:26:17)

Add MetaDeploy data models
This PR adds:

- Pydantic models to represent entities consumed or
returned by the MetaDeploy (MDP) REST API. These were not autogenerated from the API schema, however, because of the logic needed to handle mapping enums/choices from CumulusCI to MDP, as well as the generation of translation labels.

- a submodule containing functions for processing translation labels. Direct translation of the existing logic from the `metadeploy_publish` task, with those tests left in place as integration tests.


Patch:
diff --git a/cumulusci/core/metadeploy/api.py b/cumulusci/core/metadeploy/api.py
index b92e24be3..632ac071a 100644
--- a/cumulusci/core/metadeploy/api.py
+++ b/cumulusci/core/metadeploy/api.py
@@ -1,13 +1,20 @@
 """Calls to the MetaDeploy REST API."""
-
 import functools
 from typing import Optional, Union
 
+from pydantic.error_wrappers import ValidationError
 from requests.models import Response
 from requests.sessions import Session
 
 from cumulusci.core.config import ServiceConfig
 from cumulusci.core.exceptions import CumulusCIException
+from cumulusci.core.metadeploy.models import (
+    MetaDeployPlan,
+    PlanTemplate,
+    Product,
+    Slug,
+    Version,
+)
 from cumulusci.utils.http.requests_utils import safe_json_from_response
 
 
@@ -56,26 +63,40 @@ class MetaDeployAPI:
                 "ends in /rest, and that your authentication token is valid."
             ) from None
 
-    def create_plan(self, plan: dict) -> dict:
-        return self._create("/plans", plan)
+    def create_plan(self, plan: MetaDeployPlan) -> MetaDeployPlan:
+        return MetaDeployPlan.parse_obj(
+            self._create("/plans", json=plan.dict(exclude_none=True))
+        )
 
-    def create_plan_template(self, template: dict) -> dict:
-        return self._create("/plantemplates", json=template)
+    def create_plan_template(self, template: PlanTemplate) -> PlanTemplate:
+        return PlanTemplate.parse_obj(
+            self._create("/plantemplates", json=template.dict())
+        )
 
-    def create_plan_slug(self, slug: dict) -> dict:
-        return self._create("/planslug", json=slug)
+    def create_plan_slug(self, slug: Slug) -> Slug:
+        return Slug.parse_obj(
+            self._create("/planslug", json=slug.dict(include={"slug", "parent"}))
+        )
 
-    def create_version(self, version: dict) -> dict:
-        return self._create("/versions", json=version)
+    def create_version(self, version: dict) -> Version:
+        return Version.parse_obj(self._create("/versions", json=version))
 
-    def find_product(self, repo_url: str) -> Optional[dict]:
-        return self._find("/products", {"repo_url": repo_url})
+    def find_product(self, repo_url: str) -> Product:
+        try:
+            product_json = self._find("/products", {"repo_url": repo_url})
+            return Product.parse_obj(product_json)
+        except ValidationError:
+            raise CumulusCIException(
+                f"No product found in MetaDeploy with repo URL {repo_url}"
+            ) from None
 
-    def find_plan_template(self, query: dict) -> Optional[dict]:
-        return self._find("/plantemplates", query)
+    def find_plan_template(self, query: dict) -> Optional[PlanTemplate]:
+        if template_json := self._find("/plantemplates", query):
+            return PlanTemplate.parse_obj(template_json)
 
-    def find_version(self, query: dict) -> Optional[dict]:
-        return self._find("/versions", query)
+    def find_version(self, query: dict) -> Optional[Version]:
+        if version_json := self._find("/versions", query):
+            return Version.parse_obj(version_json)
 
     def update_version(self, pk: Union[int, str]) -> dict:
         response: Response = self.session.patch(
@@ -86,3 +107,25 @@ class MetaDeployAPI:
     def update_lang_translation(self, lang: str, labels: dict) -> dict:
         response: Response = self.session.patch(f"/translations/{lang}", json=labels)
         return safe_json_from_response(response)
+
+    def find_or_create_plan_template(
+        self, product_id: str, plan_template_maybe: PlanTemplate, slug: str
+    ) -> PlanTemplate:
+        query: dict = {"product": product_id, "name": plan_template_maybe.name}
+        existing_plan_template: Optional[PlanTemplate] = self.find_plan_template(query)
+
+        if existing_plan_template:
+            return existing_plan_template
+
+        plan_template = self.create_plan_template(plan_template_maybe)
+        _ = self.create_plan_slug(
+            Slug.parse_obj({"slug": slug, "parent": plan_template.url})
+        )
+        return plan_template
+
+    def find_or_create_version(
+        self, product_id: str, version_maybe: Version
+    ) -> Version:
+        query: dict = {"product": product_id, "label": version_maybe.label}
+        existing_version: Optional[Version] = self.find_version(query)
+        return existing_version or self.create_version(version_maybe.dict())
diff --git a/cumulusci/core/metadeploy/labels.py b/cumulusci/core/metadeploy/labels.py
new file mode 100644
index 000000000..8ad323c9e
--- /dev/null
+++ b/cumulusci/core/metadeploy/labels.py
@@ -0,0 +1,81 @@
+"""
+Functions for creating/modifying MetaDeploy translation labels.
+"""
+import contextlib
+import json
+import re
+from collections import defaultdict
+from pathlib import Path
+from typing import Dict, Mapping, Union
+
+LabelMap = Mapping[str, Mapping[str, str]]
+
+INSTALL_VERSION_RE = re.compile(r"^Install .*\d$")
+METADEPLOY_DIR: str = "metadeploy"
+METADEPLOY_LABELS: LabelMap = {
+    "checks": {"message": "shown if validation fails"},
+    "plan": {
+        "title": "title of installation plan",
+        "preflight_message": "shown before user starts installation (markdown)",
+        "preflight_message_additional": "shown before user starts installation (markdown)",
+        "post_install_message": "shown after successful installation (markdown)",
+        "post_install_message_additional": "shown after successful installation (markdown)",
+        "error_message": "shown after failed installation (markdown)",
+    },
+    "product": {
+        "title": "name of product",
+        "short_description": "tagline of product",
+        "description": "shown on product detail page (markdown)",
+        "click_through_agreement": "legal text shown in modal dialog",
+        "error_message": "shown after failed installation (markdown)",
+    },
+    "steps": {
+        "name": "title of installation step",
+        "description": "description of installation step",
+    },
+}
+
+
+def read_default_labels(labels_path: Union[Path, str] = METADEPLOY_DIR) -> dict:
+    """Load existing English labels from disk."""
+    labels: dict = defaultdict(dict)
+    with contextlib.suppress(FileNotFoundError):
+        labels_file: Path = Path(labels_path, "labels_en.json")
+        labels.update(json.loads(labels_file.read_text()))
+    return labels
+
+
+def read_label_files(labels_path: Union[Path, str], slug: str) -> Dict[str, dict]:
+    """
+    Load all tranlations from disk, returning prefixed labels keyed by language.
+
+    Keyword Arguments:
+    labels_path -- Path containing files to read
+    slug        -- Prepended to each label's key
+    """
+    prefixed_labels = {}
+    for path in Path(labels_path).glob("*.json"):
+        lang = path.stem.split("_")[-1].lower()
+        if lang in ("en", "en-us"):
+            continue
+        orig_labels = json.loads(path.read_text())
+        prefixed_labels[lang] = {
+            f"{slug}:{context}": labels for context, labels in orig_labels.items()
+        }
+    return prefixed_labels
+
+
+def save_default_labels(labels_path: Union[Path, str], labels_to_save: dict):
+    """Save updates to English labels."""
+    with contextlib.suppress(FileNotFoundError):
+        labels_file: Path = Path(labels_path, "labels_en.json")
+        labels_file.write_text(f"{json.dumps(labels_to_save, indent=4)}\n")
+
+
+def update_step_labels(steps, labels_to_update):
+    """Mutates labels_to_update
+    Keyword Arguments:
+    """
+    for step in steps:
+        labels_to_update["steps"].update(step.get_labels())
+        labels_to_update["checks"].update(step.task_config.get_labels())
diff --git a/cumulusci/core/metadeploy/models.py b/cumulusci/core/metadeploy/models.py
new file mode 100644
index 000000000..2cb78d7e6
--- /dev/null
+++ b/cumulusci/core/metadeploy/models.py
@@ -0,0 +1,247 @@
+"""MetaDeploy domain models"""
+
+import enum
+from typing import ChainMap, Dict, List, Literal, Mapping, Optional
+
+from pydantic import BaseModel, root_validator, validator
+from pydantic.fields import Field
+from pydantic.networks import AnyUrl
+from pydantic.types import NonNegativeInt
+
+from cumulusci.core.metadeploy.labels import (
+    INSTALL_VERSION_RE,
+    METADEPLOY_DIR,
+    METADEPLOY_LABELS,
+)
+
+
+def labels_to_field_descriptions(labels: Mapping[str, str]) -> Dict[str, dict]:
+    """Convert a LabelMap to pydantic field descriptions"""
+    return {
+        field: {"description": description} for field, description in labels.items()
+    }
+
+
+def labels_for_model(model_type: str, model_instance) -> Dict[str, dict]:
+    return {
+        name: {
+            "message": value,
+            "description": METADEPLOY_LABELS[model_type][name],
+        }
+        for name, value in model_instance
+        if name in METADEPLOY_LABELS[model_type]
+    }
+
+
+class PreflightAction(str, enum.Enum):
+    WARN = "warn"
+    ERROR = "error"
+    SKIP = "skip"
+    OPTIONAL = "optional"
+    HIDE = "hide"
+
+
+class SupportedOrgs(str, enum.Enum):
+    BOTH = "Both"
+    PERSISTENT = "Persistent"
+    SCRATCH = "Scratch"
+
+
+class MetaDeployModel(BaseModel):
+    id: Optional[str]
+    url: Optional[AnyUrl]
+
+
+class FrozenModel(BaseModel):
+    class Config:
+        allow_mutation = False
+
+
+class PreflightCheck(BaseModel):
+    when: Optional[str] = None
+    action: Optional[PreflightAction] = None
+    message: Optional[str] = None
+
+    class Config:
+        fields = {"message": {"description": "shown if validation fails"}}
+
+    def get_labels(self) -> dict:
+        if msg := self.message:
+            return {
+                msg: {
+                    "message": msg,
+                    "description": METADEPLOY_LABELS["checks"]["message"],
+                }
+            }
+        return {}
+
+
+class FrozenSpec(FrozenModel):
+    github: AnyUrl
+    commit: str
+    description: str
+
+
+class FrozenTaskConfig(FrozenModel):
+    options: dict
+    checks: List[PreflightCheck] = []
+
+    def get_labels(self):
+        labels = {}
+        if self.checks:
+            check_labels: List[dict] = [check.get_labels() for check in self.checks]
+            labels = dict(ChainMap(*check_labels))
+        return labels
+
+
+class FrozenStep(FrozenModel):
+    is_recommended: bool = True
+    is_required: bool = True
+    kind: Literal["data", "managed", "metadata", "onetime", "other"] = "other"
+    description: Optional[str] = None
+    name: str
+    path: str
+    source: Optional[FrozenSpec]
+    step_num: str
+    task_class: str
+    task_config: FrozenTaskConfig
+    url: Optional[AnyUrl] = None
+
+    def get_labels(self) -> dict:
+        # avoid separate labels for installing each package
+        name = (
+            "Install {product} {version}"
+            if INSTALL_VERSION_RE.match(self.name)
+            else self.name
+        )
+        labels = {
+            name: {
+                "message": name,
+                "description": METADEPLOY_LABELS["steps"]["name"],
+            },
+        }
+        if description := self.description:
+            labels[description] = {
+                "message": description,
+                "description": METADEPLOY_LABELS["steps"]["description"],
+            }
+        return labels
+
+
+class PublisherOptions(FrozenModel):
+    tag: Optional[str] = None
+    commit: Optional[str] = None
+    plan: Optional[str] = None
+    dry_run: bool = False
+    publish: bool = False
+    labels_path: str = METADEPLOY_DIR
+
+    @root_validator
+    def check_tag_or_commit(cls, values):
+        if not values.get("tag") and not values.get("commit"):
+            raise ValueError("You must specify either the tag or commit option.")
+        return values
+
+    @validator("publish")
+    def check_publish(cls, publish, values):
+        return not values["dry_run"] and publish
+
+
+class Product(MetaDeployModel):
+    title: str
+    short_description: str
+    description: str
+    click_through_agreement: str
+    error_message: str
+    slug: str
+    color: Optional[str]
+    image: Optional[str]
+    icon_url: Optional[str]
+    slds_icon_category: Optional[
+        Literal["", "action", "custom", "doctype", "standard", "utility"]
+    ]
+    slds_icon_name: Optional[str]
+    repo_url: Optional[str]
+    is_listed: Optional[str]
+    order_key: Optional[str]
+    layout: Optional[Literal["Default", "Card"]]
+    visible_to: Optional[str]
+    category: Optional[str]
+
+    def get_labels(self) -> Dict[str, dict]:
+        return labels_for_model("product", self)
+
+
+class Slug(MetaDeployModel):
+    parent: AnyUrl
+    is_active: Optional[bool] = None
+    slug: str = Field(..., max_length=50)
+
+
+class Version(MetaDeployModel):
+    id: Optional[str] = None
+    url: Optional[str] = None
+    label: str
+    is_production: bool = False
+    is_listed: bool = False
+    commit_ish: str
+    product: AnyUrl
+
+
+class PlanTemplate(MetaDeployModel):
+    preflight_message: str = ""
+    post_install_message: str = ""
+    error_message: str = ""
+    name: Optional[str] = Field(None, max_length=100)
+    product: AnyUrl
+    regression_test_opt_out: bool = False
+    preflight_checks: List[PreflightCheck] = Field([], alias="checks")
+
+
+class MetaDeployPlan(MetaDeployModel):
+    id: Optional[str] = None
+    commit_ish: Optional[str] = None
+    is_listed: bool = True
+    order_key: Optional[NonNegativeInt] = None
+    plan_template: Optional[AnyUrl] = None
+    post_install_message: str = Field("", exclude=True)
+    preflight_checks: List[PreflightCheck] = Field([], alias="checks")
+    preflight_message: str = Field("", exclude=True)
+    slug: Optional[str]
+    steps: List[FrozenStep] = []
+    supported_orgs: SupportedOrgs = SupportedOrgs.PERSISTENT
+    tier: Literal["primary", "secondary", "additional"]
+    title: str
+    url: Optional[AnyUrl] = None
+    version: Optional[AnyUrl] = None
+    visible_to: Optional[str]
+
+    class Config:
+        fields = labels_to_field_descriptions(METADEPLOY_LABELS["plan"])
+
+    @validator("supported_orgs", pre=True)
+    def map_supported_orgs(cls, input):
+        """Given a list of plan.allowed_org_providers return the value that
+        corresponds to the `supported_orgs` field on the MetaDeploy Plan
+        """
+        if not input:
+            return SupportedOrgs.PERSISTENT
+        try:
+            return SupportedOrgs(input)
+        except ValueError:
+            lookup: Mapping[frozenset, str] = {
+                frozenset(["user"]): SupportedOrgs.PERSISTENT,
+                frozenset(["devhub"]): SupportedOrgs.SCRATCH,
+                frozenset(["devhub", "user"]): SupportedOrgs.BOTH,
+            }
+            return lookup[frozenset(input)]
+
+    def get_labels(self) -> dict:
+        labels: dict = {f"plan:{self.slug}": labels_for_model("plan", self)}
+
+        if self.preflight_checks:
+            check_labels: List[dict] = [
+                check.get_labels() for check in self.preflight_checks
+            ]
+            labels["checks"] = dict(ChainMap(*check_labels))
+        return labels
diff --git a/cumulusci/tasks/metadeploy.py b/cumulusci/tasks/metadeploy.py
index 9c4202578..b86da29e2 100644
--- a/cumulusci/tasks/metadeploy.py
+++ b/cumulusci/tasks/metadeploy.py
@@ -1,24 +1,34 @@
-import contextlib
 import json
-import re
 from pathlib import Path
-from typing import List, Optional, Union
+from zipfile import ZipFile
 
 import requests
 
-from cumulusci.core.config import BaseProjectConfig, FlowConfig, TaskConfig
-from cumulusci.core.exceptions import CumulusCIException, TaskOptionsError
-from cumulusci.core.flowrunner import FlowCoordinator
+from cumulusci.core.config import BaseProjectConfig
+from cumulusci.core.exceptions import TaskOptionsError
 from cumulusci.core.github import get_tag_by_name
 from cumulusci.core.metadeploy.api import MetaDeployAPI
+from cumulusci.core.metadeploy.labels import (
+    METADEPLOY_DIR,
+    read_default_labels,
+    read_label_files,
+    save_default_labels,
+    update_step_labels,
+)
+from cumulusci.core.metadeploy.models import (
+    FrozenStep,
+    MetaDeployPlan,
+    PlanTemplate,
+    Product,
+    Version,
+)
+from cumulusci.core.metadeploy.plans import get_frozen_steps
 from cumulusci.core.tasks import BaseTask
 from cumulusci.core.utils import process_bool_arg
-from cumulusci.utils import cd, download_extract_github, temporary_dir
+from cumulusci.utils import download_extract_github, temporary_dir
 from cumulusci.utils.http.requests_utils import safe_json_from_response
 from cumulusci.utils.yaml.cumulusci_yml import Plan
 
-INSTALL_VERSION_RE = re.compile(r"^Install .*\d$")
-
 
 class BaseMetaDeployTask(BaseTask):
     """Base class for tasks that talk to MetaDeploy's API."""
@@ -76,21 +86,20 @@ class Publish(BaseMetaDeployTask):
             self.options.get("publish") or False
         )
         self.tag = self.options.get("tag")
-        self.commit = self.options.get("commit")
+        self.commit: str = self.options.get("commit")
         if not self.tag and not self.commit:
             raise TaskOptionsError("You must specify either the tag or commit option.")
-        self.labels_path = self.options.get("labels_path", "metadeploy")
+        self.labels_path = self.options.get("labels_path", METADEPLOY_DIR)
         Path(self.labels_path).mkdir(parents=True, exist_ok=True)
 
         if plan_name := self.options.get("plan"):
-            plan_configs = {
+            self.plan_configs = {
                 plan_name: self.project_config.lookup(f"plans__{plan_name}")
             }
-            self.plan_configs = plan_configs
         else:
             self.plan_configs = self.project_config.plans
 
-        self._load_labels()
+        self.labels = read_default_labels(self.labels_path)
 
     def _run_task(self):
         repo_owner = self.project_config.repo_owner
@@ -99,25 +108,13 @@ class Publish(BaseMetaDeployTask):
 
         # Find or create Version
         product = self._find_product(repo_url)
-        if not product:
-            raise CumulusCIException(
-                f"No product found in MetaDeploy with repo URL {repo_url}"
-            )
+
         if not self.dry_run:
             version = self._find_or_create_version(product)
             if self.labels_path and "slug" in product:
-                self._publish_labels(product["slug"])
+                self._publish_labels(product.slug)
 
-        # Check out the specified tag
-        gh = self.project_config.get_github_api()
-        repo = gh.repository(repo_owner, repo_name)
-        if self.tag:
-            tag = get_tag_by_name(repo, self.tag)
-            self.commit = tag.object.sha
-        self.logger.info(
-            f"Downloading commit {self.commit} of {repo.full_name} from GitHub"
-        )
-        zf = download_extract_github(gh, repo_owner, repo_name, ref=self.commit)
+        zf: ZipFile = self._checkout_tag(repo_owner, repo_name)
         with temporary_dir() as project_dir:
             zf.extractall(project_dir)
             project_config = BaseProjectConfig(
@@ -135,247 +132,118 @@ class Publish(BaseMetaDeployTask):
 
             # Create each plan
             for plan_name, plan_config in self.plan_configs.items():
-                self._add_plan_labels(
-                    plan_name=plan_name,
-                    plan_config=plan_config,
-                )
-
                 steps = self._freeze_steps(project_config, plan_config)
+                plan: MetaDeployPlan = MetaDeployPlan.parse_obj(
+                    dict(plan_config, steps=steps)
+                )
+                self.labels.update(plan.get_labels())
 
                 if not self.dry_run:
                     self._publish_plan(product, version, plan_name, plan_config, steps)
 
             # Update version to set is_listed=True
             if self.publish:
-                self.api.update_version(version["id"])
-                self.logger.info(f"Published Version {version['url']}")
+                self.api.update_version(version.id)
+                self.logger.info(f"Published Version {version.url}")
 
         # Save labels
-        self._save_labels()
+        self.logger.info(f"Updating labels in {self.labels_path}")
+        save_default_labels(self.labels_path, self.labels)
+
+    def _checkout_tag(self, repo_owner, repo_name) -> ZipFile:
+        # Check out the specified tag
+        gh = self.project_config.get_github_api()
+        repo = gh.repository(repo_owner, repo_name)
+        if self.tag:
+            tag = get_tag_by_name(repo, self.tag)
+            self.commit = tag.object.sha
+        self.logger.info(
+            f"Downloading commit {self.commit} of {repo.full_name} from GitHub"
+        )
+        return download_extract_github(gh, repo_owner, repo_name, ref=self.commit)
 
     def _freeze_steps(self, project_config, plan_config) -> list:
-        steps = plan_config["steps"]
-        flow_config = FlowConfig(plan_config)
-        flow_config.project_config = project_config
-        flow = FlowCoordinator(project_config, flow_config)
-        steps = []
-        for step in flow.steps:
-            if step.skip:
-                continue
-            with cd(step.project_config.repo_root):
-                task = step.task_class(
-                    step.project_config,
-                    TaskConfig(step.task_config),
-                    name=step.task_name,
-                )
-                steps.extend(task.freeze(step))
+        steps = get_frozen_steps(project_config, plan_config)
         self.logger.debug("Prepared steps:\n" + json.dumps(steps, indent=4))
 
-        self._add_step_labels(steps)
+        steps = [FrozenStep.parse_obj(step) for step in steps]
+        update_step_labels(steps, self.labels)
         return steps
 
-    def _add_plan_labels(self, plan_name, plan_config) -> None:
-        """Add labels for plans, steps, and checks."""
-        self._add_labels(
-            plan_config,
-            f"plan:{plan_name}",
-            {
-                "title": "title of installation plan",
-                "preflight_message": "shown before user starts installation (markdown)",
-                "preflight_message_additional": "shown before user starts installation (markdown)",
-                "post_install_message": "shown after successful installation (markdown)",
-                "post_install_message_additional": "shown after successful installation (markdown)",
-                "error_message": "shown after failed installation (markdown)",
-            },
-        )
-        self._add_check_labels(plan_config)
-
-    def _add_step_labels(self, steps) -> None:
-        for step in steps:
-            # avoid separate labels for installing each package
-            if INSTALL_VERSION_RE.match(step["name"]):
-                self._add_label(
-                    "steps",
-                    "Install {product} {version}",
-                    "title of installation step",
-                )
-            else:
-                self._add_label("steps", step["name"], "title of installation step")
-            self._add_label(
-                "steps",
-                step.get("description"),
-                "description of installation step",
-            )
-            self._add_check_labels(step["task_config"])
-
-    def _add_check_labels(self, config) -> None:
-        for check in config.get("checks", []):
-            self._add_label("checks", check.get("message"), "shown if validation fails")
-
     def _publish_plan(self, product, version, plan_name, plan_config, steps):
         plan_template = self._find_or_create_plan_template(
             product, plan_name, plan_config
         )
 
-        allowed_org_providers = self._get_allowed_org_providers(plan_name)
-        supported_orgs = self._convert_org_providers_to_plan_equivalent(
-            allowed_org_providers
-        )
-
-        plan_json = {
-            "is_listed": plan_config.get("is_listed", True),
-            "plan_template": plan_template["url"],
-            "post_install_message_additional": plan_config.get(
-                "post_install_message_additional", ""
-            ),
-            "preflight_message_additional": plan_config.get(
-                "preflight_message_additional", ""
-            ),
-            "steps": steps,
-            "supported_orgs": supported_orgs,
-            "tier": plan_config["tier"],
-            "title": plan_config["title"],
-            "version": version["url"],
+        parsed_plan = Plan.parse_obj(self.project_config.config["plans"][plan_name])
+        plan_json = MetaDeployPlan(
+            plan_template=plan_template.url,
+            is_listed=parsed_plan.is_listed,
+            post_install_message=plan_config.get("post_install_message", ""),
+            preflight_message=plan_config.get("post_install_message", ""),
+            tier=parsed_plan.tier,
+            title=parsed_plan.title,
+            steps=steps,
+            supported_orgs=parsed_plan.allowed_org_providers,
+            version=version.url,
             # Use same AllowedList as the product, if any
-            "visible_to": product.get("visible_to"),
-        }
-        if plan_config.get("checks"):
-            plan_json["preflight_checks"] = plan_config["checks"]
+            visible_to=product.visible_to,
+            preflight_checks=plan_config.get("checks"),
+        )
 
         # Create Plan
         plan = self.api.create_plan(plan=plan_json)
-        self.logger.info(f"Created Plan {plan['url']}")
-
-    def _get_allowed_org_providers(self, plan_name: str) -> List[str]:
-        "Validates and returns the org providers for a given plan"
-        plan = Plan.parse_obj(self.project_config.config["plans"][plan_name])
-        return plan.allowed_org_providers
-
-    def _convert_org_providers_to_plan_equivalent(
-        self, providers: Union[None, List[str]]
-    ) -> str:
-        """Given a list of plan.allowed_org_providers return the value that
-        corresponds `supported_orgs` field on the Plan model in MetaDeploy
-        """
-        if not providers or providers == ["user"]:
-            org_providers = "Persistent"
-        elif "user" in providers and "devhub" in providers:
-            org_providers = "Both"
-        elif providers == ["devhub"]:
-            org_providers = "Scratch"
-
-        return org_providers
-
-    def _find_product(self, repo_url):
-        product = self.api.find_product(repo_url)
-        self._add_labels(
-            product,
-            "product",
-            {
-                "title": "name of product",
-                "short_description": "tagline of product",
-                "description": "shown on product detail page (markdown)",
-                "click_through_agreement": "legal text shown in modal dialog",
-                "error_message": "shown after failed installation (markdown)",
-            },
-        )
+        self.logger.info(f"Created Plan {plan.url}")
+
+    def _find_product(self, repo_url) -> Product:
+        product: Product = self.api.find_product(repo_url)
+        self.labels["product"].update(product.get_labels())
         return product
 
     def _find_or_create_version(self, product):
         """Create a Version in MetaDeploy if it doesn't already exist"""
-        if self.tag:
-            label = self.project_config.get_version_for_tag(self.tag)
-        else:
-            label = self.commit
-
-        version: Optional[dict] = self.api.find_version(
-            query={"product": product["id"], "label": label}
+        label: str = self._get_label()
+
+        version_to_create = Version(
+            product=product.url,
+            label=label,
+            is_production=True,
+            commit_ish=self.tag or self.commit,
+            is_listed=False,
+        )
+        version: Version = self.api.find_or_create_version(
+            product.id, version_to_create
         )
-        verb: str = "Found" if version else "Created"
-
-        if not version:
-            version_dict = {
-                "product": product["url"],
-                "label": label,
-                "description": self.options.get("description", ""),
-                "is_production": True,
-                "commit_ish": self.tag or self.commit,
-                "is_listed": False,
-            }
-            version = self.api.create_version(version_dict)
 
-        self.logger.info(f"{verb} {version['url']}")
+        self.logger.info(f"Found or created {version.url}")
         return version
 
+    def _get_label(self) -> str:
+        if self.tag:
+            return self.project_config.get_version_for_tag(self.tag)
+        else:
+            return self.commit
+
     def _find_or_create_plan_template(self, product, plan_name, plan_config):
-        plantemplate = self.api.find_plan_template(
-            {"product": product["id"], "name": plan_name}
+        template_to_create: PlanTemplate = PlanTemplate(
+            name=plan_name,
+            product=product.url,
+            preflight_message=plan_config.get("preflight_message", ""),
+            post_install_message=plan_config.get("post_install_message", ""),
+            error_message=plan_config.get("error_message", ""),
         )
-        verb: str = "Found" if plantemplate else "Created"
-
-        if not plantemplate:
-            plantemplate = self.api.create_plan_template(
-                {
-                    "name": plan_name,
-                    "product": product["url"],
-                    "preflight_message": plan_config.get("preflight_message", ""),
-                    "post_install_message": plan_config.get("post_install_message", ""),
-                    "error_message": plan_config.get("error_message", ""),
-                }
-            )
-            planslug = self.api.create_plan_slug(
-                {"slug": plan_config["slug"], "parent": plantemplate["url"]}
-            )
-            self.logger.info(f"Created {planslug['url']}")
-        self.logger.info(f"{verb} {plantemplate['url']}")
-        return plantemplate
 
-    def _load_labels(self):
-        """Load existing English labels."""
-        try:
-            labels_path: Path = Path(self.labels_path, "labels_en.json")
-            self.labels: dict = json.loads(labels_path.read_text())
-        except FileNotFoundError:
-            self.labels: dict = {}
-
-    def _add_labels(self, obj, category, fields):
-        """Add specified fields from obj to a label category."""
-        for name, description in fields.items():
-            if text := obj.get(name):
-                if category not in self.labels:
-                    self.labels[category] = {}
-                label = {"message": text, "description": description}
-                self.labels[category][name] = label
-
-    def _add_label(self, category, text, description):
-        """Add a single label to a label category."""
-        if not text:
-            return
-        if category not in self.labels:
-            self.labels[category] = {}
-        label = {"message": text, "description": description}
-        self.labels[category][text] = label
-
-    def _save_labels(self):
-        """Save updates to English labels."""
-        with contextlib.suppress(FileNotFoundError):
-            labels_path: Path = Path(self.labels_path, "labels_en.json")
-            self.logger.info(f"Updating labels in {labels_path}")
-            labels_path.write_text(json.dumps(self.labels, indent=4))
+        plantemplate: PlanTemplate = self.api.find_or_create_plan_template(
+            product.id, template_to_create, plan_config["slug"]
+        )
+        self.logger.info(f"Found or created {plantemplate.url}")
+        return plantemplate
 
     def _publish_labels(self, slug):
         """Publish labels in all languages to MetaDeploy."""
-        for path in Path(self.labels_path).glob("*.json"):
-            lang = path.stem.split("_")[-1].lower()
-            if lang in ("en", "en-us"):
-                continue
-            orig_labels = json.loads(path.read_text())
-            prefixed_labels = {
-                f"{slug}:{context}": labels for context, labels in orig_labels.items()
-            }
-
+        for lang, label in read_label_files(self.labels_path, slug).items():
             self.logger.info(f"Updating {lang} translations")
             try:
-                self.api.update_lang_translation(lang, prefixed_labels)
+                self.api.update_lang_translation(lang, label)
             except requests.exceptions.HTTPError as err:
                 self.logger.warning(f"Could not update {lang} translation: {err}")


Test patch:
diff --git a/cumulusci/core/metadeploy/tests/conftest.py b/cumulusci/core/metadeploy/tests/conftest.py
new file mode 100644
index 000000000..0c657871e
--- /dev/null
+++ b/cumulusci/core/metadeploy/tests/conftest.py
@@ -0,0 +1,95 @@
+#!/usr/bin/env python3
+import pytest
+
+from cumulusci.core.metadeploy.models import (
+    MetaDeployPlan,
+    PlanTemplate,
+    Product,
+    Version,
+)
+
+
+@pytest.fixture
+def product_dict():
+    return {
+        "id": "abcdef",
+        "url": "https://EXISTING_PRODUCT",
+        "title": "Education Data Architecture (EDA)",
+        "short_description": "The Foundation for the Connected Campus",
+        "description": "## Welcome to the EDA installer!",
+        "click_through_agreement": "Ladies and Gentlemen of the jury, I'm just a Caveman.",
+        "error_message": "",
+        "slug": "existing",
+    }
+
+
+@pytest.fixture
+def version_dict():
+    return {
+        "url": "http://EXISTING_VERSION",
+        "id": "OkAgPpL",
+        "label": "1.0",
+        "created_at": "2022-06-30T21:15:15.390046Z",
+        "is_production": True,
+        "commit_ish": "release/1.0",
+        "is_listed": True,
+        "product": "http://localhost:8080/admin/rest/products/abcdef",
+    }
+
+
+@pytest.fixture
+def plantemplate_dict():
+    return {
+        "url": "http://localhost:8080/admin/rest/plantemplates/1",
+        "id": "1",
+        "preflight_message": "Preflight message consists of generic product message and step pre-check info — run in one operation before the install begins. Preflight includes the name of what is being installed. Lorem Ipsum has been the industry's standard dummy text ever since the 1500s.",
+        "post_install_message": "Success! You installed it.",
+        "error_message": "",
+        "name": "Full Install for Product With Useful Data, Version 0.2.0",
+        "regression_test_opt_out": False,
+        "product": "http://localhost:8080/admin/rest/products/GLN6Ppx",
+    }
+
+
+@pytest.fixture
+def simple_plan_dict():
+    return {
+        "url": "http://localhost:8080/admin/rest/plans/GLN6Ppx",
+        "id": "GLN6Ppx",
+        "steps": [],
+        "title": "Full Install",
+        "preflight_message_additional": "",
+        "post_install_message_additional": "",
+        "commit_ish": None,
+        "order_key": 0,
+        "tier": "primary",
+        "is_listed": True,
+        "preflight_checks": [],
+        "supported_orgs": "Persistent",
+        "org_config_name": "release",
+        "scratch_org_duration_override": None,
+        "created_at": "2022-06-30T21:15:11.629638Z",
+        "visible_to": None,
+        "plan_template": "http://localhost:8080/admin/rest/plantemplates/1",
+        "version": "http://localhost:8080/admin/rest/versions/GLN6Ppx",
+    }
+
+
+@pytest.fixture
+def product_model(product_dict):
+    return Product.parse_obj(product_dict)
+
+
+@pytest.fixture
+def version_model(version_dict):
+    return Version.parse_obj(version_dict)
+
+
+@pytest.fixture
+def plantemplate_model(plantemplate_dict):
+    return PlanTemplate.parse_obj(plantemplate_dict)
+
+
+@pytest.fixture
+def plan_model(simple_plan_dict):
+    return MetaDeployPlan.parse_obj(simple_plan_dict)
diff --git a/cumulusci/core/metadeploy/tests/test_api.py b/cumulusci/core/metadeploy/tests/test_api.py
index f4def5e6a..fd70c367b 100644
--- a/cumulusci/core/metadeploy/tests/test_api.py
+++ b/cumulusci/core/metadeploy/tests/test_api.py
@@ -7,6 +7,13 @@ from responses import matchers
 from cumulusci.core.config import ServiceConfig
 from cumulusci.core.exceptions import CumulusCIException
 from cumulusci.core.metadeploy.api import MetaDeployAPI, make_api_session
+from cumulusci.core.metadeploy.models import (
+    MetaDeployPlan,
+    PlanTemplate,
+    Product,
+    Slug,
+    Version,
+)
 
 pytestmark = pytest.mark.metadeploy
 DEFAULT_REST_URL: str = "https://metadeploy.example.com/api/rest"
@@ -56,15 +63,17 @@ def test_make_api(auth_header_matcher):
 
 def test_find_product(default_api_client, auth_header_matcher):
     params = {"repo_url": "https://api.github.com/repos/TestOwner/TestRepo"}
-    payload = {
-        "data": [
-            {
-                "id": "abcdef",
-                "url": "https://EXISTING_PRODUCT",
-                "slug": "existing",
-            }
-        ]
+    payload_json = {
+        "id": "abcdef",
+        "url": "https://EXISTING_PRODUCT",
+        "title": "Education Data Architecture (EDA)",
+        "short_description": "The Foundation for the Connected Campus",
+        "description": "## Welcome to the EDA installer!",
+        "click_through_agreement": "Ladies and Gentlemen of the jury, I'm just a Caveman.",
+        "error_message": "",
+        "slug": "existing",
     }
+    payload = {"data": [payload_json]}
 
     with responses.RequestsMock() as rsps:
         rsps.add(
@@ -74,8 +83,8 @@ def test_find_product(default_api_client, auth_header_matcher):
             json=payload,
         )
 
-        product: dict = default_api_client.find_product(params["repo_url"])
-    assert product == payload["data"][0]
+        product: Product = default_api_client.find_product(params["repo_url"])
+    assert product == Product.parse_obj(payload_json)
 
 
 def test_find_product__api_err(default_api_client, auth_header_matcher):
@@ -94,36 +103,40 @@ def test_find_product__api_err(default_api_client, auth_header_matcher):
             default_api_client.find_product(params["repo_url"])
 
 
-def test_create_plan(default_api_client, auth_header_matcher):
-    payload = {"url": f"{DEFAULT_REST_URL}/plans/1"}
-    plan = {
-        "is_listed": True,
-        "plan_template": "https://foo.example.com/1",
-        "post_install_message_additional": "",
-        "preflight_message_additional": "",
-        "steps": [],
-        "supported_orgs": "persistent",
-        "tier": "primary",
-        "title": "Install Tubulator 1.0",
-        "version": f"{DEFAULT_REST_URL}/version/1",
-        "visible_to": None,
-    }
+def test_create_plan(default_api_client, auth_header_matcher, simple_plan_dict):
+    plan = MetaDeployPlan(
+        is_listed=True,
+        plan_template="https://foo.example.com/1",
+        post_install_message_additional="",
+        preflight_message_additional="",
+        steps=[],
+        supported_orgs="Persistent",
+        tier="primary",
+        title="Install Tubulator 1.0",
+        version=f"{DEFAULT_REST_URL}/version/1",
+        visible_to=None,
+    )
 
     with responses.RequestsMock() as rsps:
         rsps.add(
             method=responses.POST,
             url=f"{DEFAULT_REST_URL}/plans",
-            match=(auth_header_matcher, matchers.json_params_matcher(plan)),
+            match=(
+                auth_header_matcher,
+                matchers.json_params_matcher(plan.dict(exclude_none=True)),
+            ),
             status=201,
-            json=payload,
+            json=simple_plan_dict,
         )
         result = default_api_client.create_plan(plan)
-    assert result == payload
+    assert result == MetaDeployPlan.parse_obj(simple_plan_dict)
 
 
-def test_find_version(default_api_client, auth_header_matcher):
+def test_find_version(
+    default_api_client, auth_header_matcher, version_dict, version_model
+):
     query = {"product": "bDcaVz", "label": "1.0"}
-    payload = {"data": [{"url": "http://EXISTING_VERSION"}]}
+    payload = {"data": [version_dict]}
 
     with responses.RequestsMock() as rsps:
         rsps.add(
@@ -135,7 +148,7 @@ def test_find_version(default_api_client, auth_header_matcher):
 
         result = default_api_client.find_version(query)
 
-    assert result == payload["data"][0]
+    assert result == version_model
 
 
 def test_find_version_missing_none(default_api_client, auth_header_matcher):
@@ -155,7 +168,7 @@ def test_find_version_missing_none(default_api_client, auth_header_matcher):
     assert result is None
 
 
-def test_create_version(default_api_client, auth_header_matcher):
+def test_create_version(default_api_client, auth_header_matcher, version_dict):
     version = {
         "product": f"{DEFAULT_REST_URL}/product/foo",
         "label": "1.0",
@@ -164,7 +177,8 @@ def test_create_version(default_api_client, auth_header_matcher):
         "commit_ish": "tag_or_sha",
         "is_listed": False,
     }
-    payload = {"url": f"http://{DEFAULT_REST_URL}/versions/1", "id": 1}
+    version_dict.update({"url": f"http://{DEFAULT_REST_URL}/versions/1", "id": 1})
+    expected = Version.parse_obj(version_dict)
 
     with responses.RequestsMock() as rsps:
         rsps.add(
@@ -172,12 +186,12 @@ def test_create_version(default_api_client, auth_header_matcher):
             url=f"{DEFAULT_REST_URL}/versions",
             match=(auth_header_matcher, matchers.json_params_matcher(version)),
             status=201,
-            json=payload,
+            json=version_dict,
         )
 
         version = default_api_client.create_version(version)
 
-    assert version == payload
+    assert expected == version
 
 
 def test_publish_version(default_api_client, auth_header_matcher):
@@ -197,10 +211,10 @@ def test_publish_version(default_api_client, auth_header_matcher):
     assert payload == updated_version
 
 
-def test_find_plan_template(default_api_client, auth_header_matcher):
-    query = {"product": "bDcaVz", "name": "Shinra"}
-    payload = {"data": [{"url": "http://existing_template"}]}
-
+def test_find_plan_template(default_api_client, auth_header_matcher, plantemplate_dict):
+    query = {"product": "GLN6Ppx", "name": "Shinra"}
+    payload = {"data": [plantemplate_dict]}
+    plantemplate_model = PlanTemplate.parse_obj(plantemplate_dict)
     with responses.RequestsMock() as rsps:
         rsps.add(
             method=responses.GET,
@@ -211,18 +225,14 @@ def test_find_plan_template(default_api_client, auth_header_matcher):
 
         template = default_api_client.find_plan_template(query)
 
-    assert template == payload["data"][0]
+    assert template == plantemplate_model
 
 
-def test_create_plan_template(default_api_client, auth_header_matcher):
-    template = {
-        "name": "plan_name",
-        "product": "product",
-        "preflight_message": "",
-        "post_install_message": "",
-        "error_message": "oops",
-    }
-    payload = {"url": "http://EXISTING_VERSION"}
+def test_create_plan_template(
+    default_api_client, auth_header_matcher, plantemplate_dict, plantemplate_model
+):
+    template = plantemplate_model
+    payload = plantemplate_dict
 
     with responses.RequestsMock() as rsps:
         rsps.add(
@@ -235,25 +245,33 @@ def test_create_plan_template(default_api_client, auth_header_matcher):
 
         created_template = default_api_client.create_plan_template(template)
 
-    assert created_template == payload
+    assert created_template == template
 
 
 def test_create_plan_slug(default_api_client, auth_header_matcher):
-    slug = {"slug": "plan_name", "parent": "product"}
-    payload = {"url": "http://EXISTING_VERSION"}
+    slug = Slug(slug="plan_name", parent="http://metadeploy/products/product")
+
+    payload = {
+        "slug": "plan_name",
+        "parent": "http://metadeploy/products/product",
+        "url": "http://metadeploy/plantemplates/41",
+    }
 
     with responses.RequestsMock() as rsps:
         rsps.add(
             method=responses.POST,
             url=f"{DEFAULT_REST_URL}/planslug",
-            match=(auth_header_matcher, matchers.json_params_matcher(slug)),
+            match=(
+                auth_header_matcher,
+                matchers.json_params_matcher(slug.dict(exclude_none=True)),
+            ),
             status=201,
             json=payload,
         )
 
         created_slug = default_api_client.create_plan_slug(slug)
 
-    assert created_slug == payload
+    assert created_slug == Slug.parse_obj(payload)
 
 
 def test_publish_labels(default_api_client, auth_header_matcher):
diff --git a/cumulusci/core/metadeploy/tests/test_labels.py b/cumulusci/core/metadeploy/tests/test_labels.py
new file mode 100644
index 000000000..f437b8388
--- /dev/null
+++ b/cumulusci/core/metadeploy/tests/test_labels.py
@@ -0,0 +1,136 @@
+import json
+from collections import defaultdict
+from copy import deepcopy
+from pathlib import Path
+
+import pytest
+
+from cumulusci.core.metadeploy.labels import (
+    METADEPLOY_DIR,
+    METADEPLOY_LABELS,
+    read_default_labels,
+    read_label_files,
+    save_default_labels,
+    update_step_labels,
+)
+from cumulusci.core.metadeploy.models import FrozenStep
+
+pytestmark = pytest.mark.metadeploy
+
+
+def test_read_default_labels(tmp_path):
+    label_path: Path = tmp_path / METADEPLOY_DIR
+    label_path.mkdir(parents=True)
+    label_file: Path = label_path / "labels_en.json"
+    label_file.write_text(json.dumps(METADEPLOY_LABELS))
+
+    result: dict = read_default_labels(labels_path=label_path)
+    assert METADEPLOY_LABELS == result, "should match input"
+    assert {} == result["missing_category"], "should be a default dict"
+
+
+def test_read_default_labels_missing():
+    result: dict = read_default_labels(labels_path="missing_dir")
+    assert {} == result, "should be empty dict"
+    assert {} == result["missing_category"], "should be a default dict"
+
+
+def test_read_label_files(tmp_path):
+    label_path: Path = tmp_path / METADEPLOY_DIR
+    en_label_file: Path = label_path / "labels_en.json"
+    fr_label_file: Path = label_path / "labels_fr.json"
+    label_path.mkdir(parents=True)
+    label = {"message": "message", "label": "label"}
+    en_label_file.write_text(json.dumps(label))
+    fr_label_file.write_text(json.dumps(label))
+
+    expected_dict = {"fr": {"slug:message": "message", "slug:label": "label"}}
+
+    result: dict = read_label_files(label_path, "slug")
+    assert expected_dict == result
+
+
+def test_save_labels(tmp_path):
+    label_path: Path = tmp_path / METADEPLOY_DIR
+    label_path.mkdir(parents=True)
+    label_file: Path = label_path / "labels_en.json"
+    expected = json.dumps(METADEPLOY_LABELS, indent=4)
+
+    save_default_labels(label_path, METADEPLOY_LABELS)
+
+    result = label_file.read_text()
+    assert expected == result.strip()  # delete trailing newline
+
+
+def test_update_step_labels():
+    steps = [
+        FrozenStep.parse_obj(
+            {
+                "is_recommended": True,
+                "is_required": True,
+                "kind": "managed",
+                "message": None,
+                "description": None,
+                "name": "Install Test Product 1.0",
+                "path": "install_prod.install_managed",
+                "source": None,
+                "step_num": "1/2",
+                "task_class": "cumulusci.tasks.salesforce.InstallPackageVersion",
+                "task_config": {
+                    "options": {
+                        "version": "1.0",
+                        "namespace": "ns",
+                        "interactive": False,
+                        "base_package_url_format": "{}",
+                    },
+                    "checks": [],
+                },
+                "url": None,
+            }
+        ),
+        FrozenStep.parse_obj(
+            {
+                "is_recommended": True,
+                "is_required": True,
+                "kind": "other",
+                "message": None,
+                "name": "util_sleep",
+                "path": "util_sleep",
+                "step_num": "2",
+                "source": None,
+                "task_class": "cumulusci.tasks.util.Sleep",
+                "task_config": {
+                    "options": {"seconds": 5},
+                    "checks": [
+                        {
+                            "when": "False",
+                            "action": "error",
+                            "message": "Check failed 😭",
+                        },
+                        {"when": "False", "action": "error", "message": None},
+                    ],
+                },
+                "url": None,
+            }
+        ),
+    ]
+    labels_to_update = defaultdict(dict)
+    expected_dict = deepcopy(labels_to_update)
+    expected_dict["steps"] = {
+        "Install {product} {version}": {
+            "message": "Install {product} {version}",
+            "description": METADEPLOY_LABELS["steps"]["name"],
+        },
+        "util_sleep": {
+            "message": "util_sleep",
+            "description": METADEPLOY_LABELS["steps"]["name"],
+        },
+    }
+    expected_dict["checks"] = {
+        "Check failed 😭": {
+            "message": "Check failed 😭",
+            "description": METADEPLOY_LABELS["checks"]["message"],
+        }
+    }
+    update_step_labels(steps, labels_to_update)
+    assert expected_dict == labels_to_update
diff --git a/cumulusci/core/metadeploy/tests/test_models.py b/cumulusci/core/metadeploy/tests/test_models.py
new file mode 100644
index 000000000..43fa1c618
--- /dev/null
+++ b/cumulusci/core/metadeploy/tests/test_models.py
@@ -0,0 +1,212 @@
+from typing import Dict
+
+import pytest
+from pydantic import ValidationError
+
+from cumulusci.core.metadeploy.labels import METADEPLOY_LABELS
+from cumulusci.core.metadeploy.models import (
+    FrozenStep,
+    MetaDeployPlan,
+    PreflightCheck,
+    PublisherOptions,
+    SupportedOrgs,
+)
+
+pytestmark = pytest.mark.metadeploy
+
+
+@pytest.mark.parametrize(
+    "input,expected",
+    zip(
+        [
+            ["user", "devhub"],
+            ["user"],
+            ["devhub"],
+        ],
+        list(SupportedOrgs),
+    ),
+)
+def test_supported_orgs_validation_map(input, expected, simple_plan_dict):
+    del simple_plan_dict["supported_orgs"]
+    result = MetaDeployPlan(supported_orgs=input, **simple_plan_dict)
+    assert expected == result.supported_orgs
+
+
+def test_supported_orgs_validation_empty(simple_plan_dict):
+    del simple_plan_dict["supported_orgs"]
+    persistent = MetaDeployPlan(supported_orgs=None, **simple_plan_dict)
+    assert persistent.supported_orgs == SupportedOrgs.PERSISTENT
+
+    empty = MetaDeployPlan(supported_orgs=[], **simple_plan_dict)
+    assert empty.supported_orgs == SupportedOrgs.PERSISTENT
+
+
+def test_publisher_options_tag_validation():
+    with pytest.raises(
+        ValidationError, match="You must specify either the tag or commit option."
+    ):
+        PublisherOptions(plan="slug", dry_run=True, publish=True)
+
+
+def test_publisher_options_no_publish_on_dry_run():
+    opts: PublisherOptions = PublisherOptions(
+        plan="slug", dry_run=True, publish=True, tag="v2.0"
+    )
+    assert not opts.publish
+
+
+def test_product_get_labels(product_model):
+    expected: Dict[str, dict] = {
+        "short_description": {
+            "description": METADEPLOY_LABELS["product"]["short_description"],
+            "message": product_model.short_description,
+        },
+        "title": {
+            "description": METADEPLOY_LABELS["product"]["title"],
+            "message": product_model.title,
+        },
+        "description": {
+            "description": METADEPLOY_LABELS["product"]["description"],
+            "message": product_model.description,
+        },
+        "click_through_agreement": {
+            "description": METADEPLOY_LABELS["product"]["click_through_agreement"],
+            "message": product_model.click_through_agreement,
+        },
+        "error_message": {
+            "description": METADEPLOY_LABELS["product"]["error_message"],
+            "message": product_model.error_message,
+        },
+    }
+    actual = product_model.get_labels()
+    assert expected == actual
+
+
+def test_preflight_get_labels():
+    empty_check: PreflightCheck = PreflightCheck(
+        when="when 'foo' in tasks.get_stuff()", action="skip"
+    )
+    assert not empty_check.get_labels()
+
+    msg = "Scary warning message"
+    check: PreflightCheck = PreflightCheck(
+        when="when 'foo' in tasks.get_stuff()",
+        action="warn",
+        message=msg,
+    )
+    expected = {
+        msg: {
+            "message": msg,
+            "description": METADEPLOY_LABELS["checks"]["message"],
+        }
+    }
+    assert expected == check.get_labels()
+
+
+def test_plan_labels(plan_model):
+    plan_model.slug = "slug"
+    print(plan_model)
+    expected_dict = {
+        "plan:slug": {
+            "title": {
+                "message": plan_model.title,
+                "description": METADEPLOY_LABELS["plan"]["title"],
+            },
+            "post_install_message": {
+                "message": plan_model.post_install_message,
+                "description": "shown after successful installation (markdown)",
+            },
+            "preflight_message": {
+                "message": plan_model.preflight_message,
+                "description": "shown before user starts installation (markdown)",
+            },
+        },
+    }
+    labels = plan_model.get_labels()
+    assert expected_dict == labels
+
+    plan_model.preflight_checks = [
+        PreflightCheck(
+            when="when 'foo' in tasks.get_stuff()",
+            action="warn",
+            message="check_msg",
+        )
+    ]
+    expected_dict["checks"] = {
+        "check_msg": {
+            "message": "check_msg",
+            "description": METADEPLOY_LABELS["checks"]["message"],
+        }
+    }
+    labels_with_checks = plan_model.get_labels()
+    assert expected_dict == labels_with_checks
+
+
+def test_update_step_labels():
+    step: FrozenStep = FrozenStep.parse_obj(
+        {
+            "is_recommended": True,
+            "is_required": True,
+            "kind": "other",
+            "message": None,
+            "name": "util_sleep",
+            "path": "util_sleep",
+            "step_num": "2",
+            "source": None,
+            "task_class": "cumulusci.tasks.util.Sleep",
+            "task_config": {
+                "options": {"seconds": 5},
+                "checks": [
+                    {"when": "False", "action": "warn", "message": "check_msg"},
+                    {"when": "False", "action": "error", "message": None},
+                ],
+            },
+            "url": None,
+        }
+    )
+    expected_dict = {
+        "util_sleep": {
+            "message": "util_sleep",
+            "description": METADEPLOY_LABELS["steps"]["name"],
+        }
+    }
+    labels = step.get_labels()
+    assert expected_dict == labels
+
+
+def test_update_step_labels_regex():
+    step: FrozenStep = FrozenStep.parse_obj(
+        {
+            "is_recommended": True,
+            "is_required": True,
+            "kind": "managed",
+            "message": None,
+            "description": "Step description",
+            "name": "Install Test Product 1.0",
+            "path": "install_prod.install_managed",
+            "source": None,
+            "step_num": "1/2",
+            "task_class": "cumulusci.tasks.salesforce.InstallPackageVersion",
+            "task_config": {
+                "options": {
+                    "version": "1.0",
+                    "namespace": "ns",
+                    "interactive": False,
+                    "base_package_url_format": "{}",
+                },
+                "checks": [],
+            },
+        }
+    )
+    expected_dict = {
+        "Install {product} {version}": {
+            "message": "Install {product} {version}",
+            "description": METADEPLOY_LABELS["steps"]["name"],
+        },
+        "Step description": {
+            "message": "Step description",
+            "description": METADEPLOY_LABELS["steps"]["description"],
+        },
+    }
+    labels = step.get_labels()
+    assert expected_dict == labels
diff --git a/cumulusci/core/metadeploy/tests/test_plans.py b/cumulusci/core/metadeploy/tests/test_plans.py
new file mode 100644
index 000000000..bfe38f08f
--- /dev/null
+++ b/cumulusci/core/metadeploy/tests/test_plans.py
@@ -0,0 +1,95 @@
+import pytest
+
+from cumulusci.core.config import ServiceConfig
+from cumulusci.core.metadeploy.plans import get_frozen_steps
+from cumulusci.tests.util import create_project_config
+
+pytestmark = pytest.mark.metadeploy
+
+
+def test_freeze_steps():
+    project_config = create_project_config()
+    project_config.config["flows"] = {
+        "customer_org": {
+            "steps": {
+                1: {
+                    "task": "install_managed",
+                    "options": {
+                        "version": "1.0",
+                        "namespace": "ns",
+                        "name": "Test Product",
+                    },
+                },
+            }
+        }
+    }
+    plan_config = {
+        "title": "Test Install",
+        "slug": "install",
+        "tier": "primary",
+        "steps": {
+            1: {"flow": "customer_org"},
+            2: {
+                "task": "util_sleep",
+                "checks": [{"when": "False", "action": "error"}],
+            },
+        },
+        "checks": [{"when": "False", "action": "error"}],
+        "allowed_org_providers": ["user", "devhub"],
+    }
+    steps = get_frozen_steps(project_config, plan_config)
+    assert [
+        {
+            "is_required": True,
+            "kind": "managed",
+            "name": "Install Test Product 1.0",
+            "path": "customer_org.install_managed",
+            "source": None,
+            "step_num": "1/1",
+            "task_class": "cumulusci.tasks.salesforce.InstallPackageVersion",
+            "task_config": {
+                "checks": [],
+                "options": {
+                    "version": "1.0",
+                    "namespace": "ns",
+                    "interactive": False,
+                    "base_package_url_format": "{}",
+                },
+            },
+        },
+        {
+            "is_required": True,
+            "kind": "other",
+            "name": "util_sleep",
+            "path": "util_sleep",
+            "step_num": "2",
+            "source": None,
+            "task_class": "cumulusci.tasks.util.Sleep",
+            "task_config": {
+                "options": {"seconds": 5},
+                "checks": [{"when": "False", "action": "error"}],
+            },
+        },
+    ] == steps
+
+
+def test_freeze_steps__skip():
+    project_config = create_project_config()
+    project_config.keychain.set_service(
+        "metadeploy",
+        "test_alias",
+        ServiceConfig({"url": "https://metadeploy", "token": "TOKEN"}),
+    )
+    project_config.keychain.set_service(
+        "github",
+        "test_alias",
+        ServiceConfig({"username": "foo", "token": "bar", "email": "foo@example.com"}),
+    )
+    plan_config = {
+        "title": "Test Install",
+        "slug": "install",
+        "tier": "primary",
+        "steps": {1: {"task": "None"}},
+    }
+    steps = get_frozen_steps(project_config, plan_config)
+    assert steps == []
diff --git a/cumulusci/tasks/tests/test_metadeploy.py b/cumulusci/tasks/tests/test_metadeploy.py
index bda320555..d575164cd 100644
--- a/cumulusci/tasks/tests/test_metadeploy.py
+++ b/cumulusci/tasks/tests/test_metadeploy.py
@@ -13,6 +13,7 @@ import yaml
 
 from cumulusci.core.config import ServiceConfig, TaskConfig
 from cumulusci.core.exceptions import CumulusCIException, TaskOptionsError
+from cumulusci.core.metadeploy.models import Product, Version
 from cumulusci.tasks.github.tests.util_github_api import GithubApiTestMixin
 from cumulusci.tasks.metadeploy import BaseMetaDeployTask, Publish
 from cumulusci.tests.util import create_project_config
@@ -20,6 +21,82 @@ from cumulusci.tests.util import create_project_config
 pytestmark = pytest.mark.metadeploy
 
 
+@pytest.fixture
+def product_dict():
+    return {
+        "id": "abcdef",
+        "url": "https://EXISTING_PRODUCT",
+        "title": "Education Data Architecture (EDA)",
+        "short_description": "The Foundation for the Connected Campus",
+        "description": "## Welcome to the EDA installer!",
+        "click_through_agreement": "Ladies and Gentlemen of the jury, I'm just a Caveman.",
+        "error_message": "",
+        "slug": "existing",
+    }
+
+
+@pytest.fixture
+def simple_plan_dict():
+    return {
+        "url": "http://localhost:8080/admin/rest/plans/GLN6Ppx",
+        "id": "GLN6Ppx",
+        "steps": [],
+        "title": "Full Install",
+        "preflight_message_additional": "",
+        "post_install_message_additional": "",
+        "commit_ish": None,
+        "order_key": 0,
+        "tier": "primary",
+        "is_listed": True,
+        "preflight_checks": [],
+        "supported_orgs": "Persistent",
+        "org_config_name": "release",
+        "scratch_org_duration_override": None,
+        "created_at": "2022-06-30T21:15:11.629638Z",
+        "visible_to": None,
+        "plan_template": "http://localhost:8080/admin/rest/plantemplates/1",
+        "version": "http://localhost:8080/admin/rest/versions/GLN6Ppx",
+    }
+
+
+@pytest.fixture
+def version_dict():
+    return {
+        "url": "http://EXISTING_VERSION",
+        "id": "OkAgPpL",
+        "label": "1.0",
+        "created_at": "2022-06-30T21:15:15.390046Z",
+        "is_production": True,
+        "commit_ish": "release/1.0",
+        "is_listed": True,
+        "product": "http://localhost:8080/admin/rest/products/abcdef",
+    }
+
+
+@pytest.fixture
+def plantemplate_dict():
+    return {
+        "url": "http://localhost:8080/admin/rest/plantemplates/1",
+        "id": "1",
+        "preflight_message": "Preflight message consists of generic product message and step pre-check info — run in one operation before the install begins. Preflight includes the name of what is being installed. Lorem Ipsum has been the industry's standard dummy text ever since the 1500s.",
+        "post_install_message": "Success! You installed it.",
+        "error_message": "",
+        "name": "Full Install for Product With Useful Data, Version 0.2.0",
+        "regression_test_opt_out": False,
+        "product": "http://localhost:8080/admin/rest/products/GLN6Ppx",
+    }
+
+
+@pytest.fixture
+def product_model(product_dict):
+    return Product.parse_obj(product_dict)
+
+
+@pytest.fixture
+def version_model(version_dict):
+    return Version.parse_obj(version_dict)
+
+
 class TestBaseMetaDeployTask:
     maxDiff = None
 
@@ -113,7 +190,9 @@ class TestPublish(GithubApiTestMixin):
         assert "Admin API" in str(e.value)
 
     @responses.activate
-    def test_run_task(self):
+    def test_run_task(
+        self, simple_plan_dict, product_dict, version_dict, plantemplate_dict
+    ):
         project_config = create_project_config()
         project_config.config["plans"] = {
             "install": {
@@ -147,15 +226,7 @@ class TestPublish(GithubApiTestMixin):
         responses.add(
             "GET",
             "https://metadeploy/products?repo_url=https%3A%2F%2Fgithub.com%2FTestOwner%2FTestRepo",
-            json={
-                "data": [
-                    {
-                        "id": "abcdef",
-                        "url": "https://EXISTING_PRODUCT",
-                        "slug": "existing",
-                    }
-                ]
-            },
+            json={"data": [product_dict]},
         )
         responses.add(
             "GET",
@@ -237,19 +308,21 @@ class TestPublish(GithubApiTestMixin):
         responses.add(
             "POST",
             "https://metadeploy/versions",
-            json={"url": "https:/metadeploy/versions/1", "id": 1},
+            json=version_dict,
         )
         responses.add(
             "GET",
             "https://metadeploy/plantemplates?product=abcdef&name=install",
-            json={"data": [{"url": "https://metadeploy/plantemplates/1"}]},
+            json={"data": [plantemplate_dict]},
         )
         responses.add(
             "POST",
             "https://metadeploy/plans",
-            json={"url": "https://metadeploy/plans/1"},
+            json=simple_plan_dict,
+        )
+        responses.add(
+            "PATCH", f"https://metadeploy/versions/{version_dict['id']}", json={}
         )
-        responses.add("PATCH", "https://metadeploy/versions/1", json={})
 
         labels_path = tempfile.mkdtemp()
         en_labels_path = Path(labels_path, "labels_en.json")
@@ -279,17 +352,17 @@ class TestPublish(GithubApiTestMixin):
         self.maxDiff = None
         assert [
             {
+                "is_recommended": True,
                 "is_required": True,
                 "kind": "managed",
                 "name": "Install Test Product 1.0",
                 "path": "install_prod.install_managed",
-                "source": None,
                 "step_num": "1/2",
                 "task_class": "cumulusci.tasks.salesforce.InstallPackageVersion",
                 "task_config": {
                     "options": {
-                        "namespace": "ns",
                         "version": "1.0",
+                        "namespace": "ns",
                         "interactive": False,
                         "base_package_url_format": "{}",
                     },
@@ -297,11 +370,11 @@ class TestPublish(GithubApiTestMixin):
                 },
             },
             {
+                "is_recommended": True,
                 "is_required": True,
                 "kind": "metadata",
                 "name": "Update Admin Profile",
                 "path": "install_prod.config_managed.update_admin_profile",
-                "source": None,
                 "step_num": "1/3/2",
                 "task_class": "cumulusci.tasks.salesforce.ProfileGrantAllAccess",
                 "task_config": {
@@ -313,9 +386,10 @@ class TestPublish(GithubApiTestMixin):
                 },
             },
             {
-                "name": "util_sleep",
-                "kind": "other",
+                "is_recommended": True,
                 "is_required": True,
+                "kind": "other",
+                "name": "util_sleep",
                 "path": "util_sleep",
                 "step_num": "2",
                 "task_class": "cumulusci.tasks.util.Sleep",
@@ -323,17 +397,24 @@ class TestPublish(GithubApiTestMixin):
                     "options": {"seconds": 5},
                     "checks": [{"when": "False", "action": "error"}],
                 },
-                "source": None,
             },
         ] == steps
 
         labels = json.loads(en_labels_path.read_text())
         assert labels == {
             "plan:install": {
+                "post_install_message": {
+                    "message": "",
+                    "description": "shown after successful installation (markdown)",
+                },
+                "preflight_message": {
+                    "message": "",
+                    "description": "shown before user starts installation (markdown)",
+                },
                 "title": {
                     "message": "Test Install",
                     "description": "title of installation plan",
-                }
+                },
             },
             "steps": {
                 "Install {product} {version}": {
@@ -350,15 +431,38 @@ class TestPublish(GithubApiTestMixin):
                 },
             },
             "test": {"title": {}},
+            "checks": {},
+            "product": {
+                "title": {
+                    "message": "Education Data Architecture (EDA)",
+                    "description": "name of product",
+                },
+                "short_description": {
+                    "message": "The Foundation for the Connected Campus",
+                    "description": "tagline of product",
+                },
+                "description": {
+                    "message": "## Welcome to the EDA installer!",
+                    "description": "shown on product detail page (markdown)",
+                },
+                "click_through_agreement": {
+                    "message": "Ladies and Gentlemen of the jury, I'm just a Caveman.",
+                    "description": "legal text shown in modal dialog",
+                },
+                "error_message": {
+                    "message": "",
+                    "description": "shown after failed installation (markdown)",
+                },
+            },
         }
         shutil.rmtree(labels_path)
 
     @responses.activate
-    def test_find_or_create_version__already_exists(self):
+    def test_find_or_create_version__already_exists(self, product_model, version_dict):
         responses.add(
             "GET",
             "https://metadeploy/versions?product=abcdef&label=1.0",
-            json={"data": [{"url": "http://EXISTING_VERSION"}]},
+            json={"data": [version_dict]},
         )
 
         project_config = create_project_config()
@@ -379,17 +483,15 @@ class TestPublish(GithubApiTestMixin):
         task_config = TaskConfig({"options": {"tag": "release/1.0"}})
         task = Publish(project_config, task_config)
         task._init_task()
-        version = task._find_or_create_version(
-            {"url": "http://EXISTING_PRODUCT", "id": "abcdef"}
-        )
-        assert version["url"] == "http://EXISTING_VERSION"
+        version = task._find_or_create_version(product_model)
+        assert version.url == "http://EXISTING_VERSION"
 
     @responses.activate
-    def test_find_or_create_version__commit(self):
+    def test_find_or_create_version__commit(self, product_model, version_dict):
         responses.add(
             "GET",
             "https://metadeploy/versions?product=abcdef&label=abcdef",
-            json={"data": [{"url": "http://EXISTING_VERSION"}]},
+            json={"data": [version_dict]},
         )
 
         project_config = create_project_config()
@@ -410,10 +512,8 @@ class TestPublish(GithubApiTestMixin):
         task_config = TaskConfig({"options": {"commit": "abcdef"}})
         task = Publish(project_config, task_config)
         task._init_task()
-        version = task._find_or_create_version(
-            {"url": "http://EXISTING_PRODUCT", "id": "abcdef"}
-        )
-        assert version["url"] == "http://EXISTING_VERSION"
+        version = task._find_or_create_version(product_model)
+        assert version.url == "http://EXISTING_VERSION"
 
     @responses.activate
     def test_find_product__not_found(self):
@@ -471,7 +571,9 @@ class TestPublish(GithubApiTestMixin):
         assert expected_plans == task.plan_configs
 
     @responses.activate
-    def test_find_or_create_plan_template__not_found(self):
+    def test_find_or_create_plan_template__not_found(
+        self, product_model, plantemplate_dict
+    ):
         responses.add(
             "GET",
             "https://metadeploy/plantemplates?product=abcdef&name=install",
@@ -480,10 +582,16 @@ class TestPublish(GithubApiTestMixin):
         responses.add(
             "POST",
             "https://metadeploy/plantemplates",
-            json={"url": "https://NEW_PLANTEMPLATE"},
+            json=plantemplate_dict,
         )
         responses.add(
-            "POST", "https://metadeploy/planslug", json={"url": "http://NEW_PLANSLUG"}
+            "POST",
+            "https://metadeploy/planslug",
+            json={
+                "parent": plantemplate_dict["product"],
+                "slug": "install",
+                "url": "http://NEW_PLANSLUG",
+            },
         )
 
         project_config = create_project_config()
@@ -506,11 +614,11 @@ class TestPublish(GithubApiTestMixin):
         task = Publish(project_config, task_config)
         task._init_task()
         plantemplate = task._find_or_create_plan_template(
-            {"url": "https://EXISTING_PRODUCT", "id": "abcdef"},
+            product_model,
             "install",
             {"slug": "install"},
         )
-        assert plantemplate["url"] == "https://NEW_PLANTEMPLATE"
+        assert plantemplate.url == plantemplate_dict["url"]
 
     def test_freeze_steps__skip(self):
         project_config = create_project_config()
@@ -530,38 +638,3 @@ class TestPublish(GithubApiTestMixin):
         task._init_task()
         steps = task._freeze_steps(project_config, plan_config)
         assert steps == []
-
-    providers = [
-        (["user"], "Persistent"),
-        (["devhub"], "Scratch"),
-        (["user", "devhub"], "Both"),
-    ]
-
-    @pytest.mark.parametrize("org_providers,metadeploy_equivalent", providers)
-    def test_convert_org_providers_to_plan_equivalent(
-        self, org_providers, metadeploy_equivalent
-    ):
-        project_config = create_project_config()
-        project_config.keychain.set_service(
-            "metadeploy",
-            "test_alias",
-            ServiceConfig({"url": "https://metadeploy", "token": "TOKEN"}),
-        )
-
-        plan_config = {
-            "title": "Test Install",
-            "slug": "install",
-            "tier": "primary",
-            "steps": {1: {"task": "None"}},
-            "allowed_org_providers": org_providers,
-        }
-        plan_name = "test_install"
-        project_config.config["plans"] = {
-            plan_name: plan_config,
-        }
-        task_config = TaskConfig({"options": {"tag": "release/1.0"}})
-
-        task = Publish(project_config, task_config)
-        actual = task._convert_org_providers_to_plan_equivalent(org_providers)
-
-        assert actual == metadeploy_equivalent