Merge pull request #3 from git-mastery/rewrite-logic-flow

Rewrite logic flow

Author: woojiahao

.github/workflows/publish.yml

@@ -30,6 +30,10 @@ jobs:
           python -m pip install --upgrade pip
           pip install -r requirements.txt
 
+      - name: Run unit tests
+        run: |
+          python -m pytest -s -vv
+
       - name: Build binary
         run: |
           echo "__version__ = \"${GITHUB_REF_NAME}\"" > src/repo_smith/version.py

README.md

@@ -53,3 +53,9 @@ For more use cases of `repo-smith`, refer to:
 
 - [Official specification](/specification.md)
 - [Unit tests](./tests/)
+
+## FAQ
+
+### Why don't you assign every error to a constant and unit test against the constant?
+
+Suppose the constant was `X`, and we unit tested that the error value was `X`, we would not capture any nuances if the value of `X` had changed by accident.

requirements.txt

@@ -4,3 +4,4 @@ PyYAML
 types-PyYAML
 build
 twine
+pytest-cov

src/repo_smith/initialize_repo.py

@@ -1,5 +1,4 @@
 import os
-import re
 import shutil
 import tempfile
 from contextlib import contextmanager
@@ -8,22 +7,10 @@
 import yaml
 from git import Repo
 
-import repo_smith.steps.add_step
-import repo_smith.steps.bash_step
-import repo_smith.steps.branch_delete_step
-import repo_smith.steps.branch_rename_step
-import repo_smith.steps.branch_step
-import repo_smith.steps.checkout_step
-import repo_smith.steps.commit_step
-import repo_smith.steps.fetch_step
-import repo_smith.steps.file_step
-import repo_smith.steps.merge_step
-import repo_smith.steps.remote_step
 import repo_smith.steps.tag_step
 from repo_smith.clone_from import CloneFrom
 from repo_smith.spec import Spec
-from repo_smith.steps.step import Step
-from repo_smith.steps.step_type import StepType
+from repo_smith.steps.dispatcher import Dispatcher
 
 Hook: TypeAlias = Callable[[Repo], None]
 
@@ -46,7 +33,7 @@ def initialize(self, existing_path: Optional[str] = None) -> Iterator[Repo]:
             if self.__spec.clone_from is not None:
                 repo = Repo.clone_from(self.__spec.clone_from.repo_url, tmp_dir)
             else:
-                repo = Repo.init(tmp_dir)
+                repo = Repo.init(tmp_dir, initial_branch="main")
 
             for step in self.__spec.steps:
                 if step.id in self.__pre_hooks:
@@ -118,8 +105,8 @@ def __get_all_ids(self, spec: Spec) -> Set[str]:
     def __parse_spec(self, spec: Any) -> Spec:
         steps = []
 
-        for step in spec.get("initialization", {}).get("steps", []):
-            steps.append(self.__parse_step(step))
+        for step in spec.get("initialization", {}).get("steps", []) or []:
+            steps.append(Dispatcher.dispatch(step))
 
         clone_from = None
         if spec.get("initialization", {}).get("clone-from", None) is not None:
@@ -134,233 +121,6 @@ def __parse_spec(self, spec: Any) -> Spec:
             clone_from=clone_from,
         )
 
-    def __parse_step(self, step: Any) -> Step:
-        if "type" not in step:
-            raise ValueError('Missing "type" field in step.')
-
-        name = step.get("name")
-        description = step.get("description")
-        step_type = StepType.from_value(step["type"])
-        id = step.get("id")
-
-        if step_type == StepType.COMMIT:
-            if "message" not in step:
-                raise ValueError('Missing "message" field in commit step.')
-
-            return repo_smith.steps.commit_step.CommitStep(
-                name=name,
-                description=description,
-                step_type=StepType.COMMIT,
-                id=id,
-                empty=step.get("empty", False),
-                message=step["message"],
-            )
-        elif step_type == StepType.ADD:
-            if "files" not in step:
-                raise ValueError('Missing "files" field in add step.')
-
-            if step["files"] is None or step["files"] == []:
-                raise ValueError('Empty "files" list in add step.')
-
-            return repo_smith.steps.add_step.AddStep(
-                name=name,
-                description=description,
-                step_type=StepType.ADD,
-                id=id,
-                files=step["files"],
-            )
-        elif step_type == StepType.TAG:
-            if "tag-name" not in step:
-                raise ValueError('Missing "tag-name" field in tag step.')
-
-            if step["tag-name"] is None or step["tag-name"].strip() == "":
-                raise ValueError('Empty "tag-name" field in tag step.')
-
-            tag_name_regex = "^[0-9a-zA-Z-_.]*$"
-            if re.search(tag_name_regex, step["tag-name"]) is None:
-                raise ValueError(
-                    'Field "tag-name" can only contain alphanumeric characters, _, -, .'
-                )
-
-            return repo_smith.steps.tag_step.TagStep(
-                name=name,
-                description=description,
-                step_type=StepType.TAG,
-                id=id,
-                tag_name=step["tag-name"],
-                tag_message=step.get("tag-message"),
-            )
-        elif step_type == StepType.BASH:
-            if "runs" not in step:
-                raise ValueError('Missing "runs" field in bash step.')
-
-            if step["runs"] is None or step["runs"].strip() == "":
-                raise ValueError('Empty "runs" field in file step.')
-
-            return repo_smith.steps.bash_step.BashStep(
-                name=name,
-                description=description,
-                step_type=step_type,
-                id=id,
-                body=step["runs"],
-            )
-        elif step_type == StepType.BRANCH:
-            if "branch-name" not in step:
-                raise ValueError('Missing "branch-name" field in branch step.')
-
-            if step["branch-name"] is None or step["branch-name"].strip() == "":
-                raise ValueError('Empty "branch-name" field in branch step.')
-
-            return repo_smith.steps.branch_step.BranchStep(
-                name=name,
-                description=description,
-                step_type=step_type,
-                id=id,
-                branch_name=step["branch-name"],
-            )
-        elif step_type == StepType.BRANCH_RENAME:
-            if "branch-name" not in step:
-                raise ValueError('Missing "branch-name" field in branch-rename step.')
-
-            if step["branch-name"] is None or step["branch-name"].strip() == "":
-                raise ValueError('Empty "branch-name" field in branch-rename step.')
-
-            if "new-name" not in step:
-                raise ValueError('Missing "new-name" field in branch-rename step.')
-
-            if step["new-name"] is None or step["new-name"].strip() == "":
-                raise ValueError('Empty "new-name" field in branch-rename step.')
-
-            return repo_smith.steps.branch_rename_step.BranchRenameStep(
-                name=name,
-                description=description,
-                step_type=step_type,
-                id=id,
-                original_branch_name=step["branch-name"],
-                target_branch_name=step["new-name"],
-            )
-        elif step_type == StepType.BRANCH_DELETE:
-            if "branch-name" not in step:
-                raise ValueError('Missing "branch-name" field in branch step.')
-
-            if step["branch-name"] is None or step["branch-name"].strip() == "":
-                raise ValueError('Empty "branch-name" field in branch step.')
-
-            return repo_smith.steps.branch_delete_step.BranchDeleteStep(
-                name=name,
-                description=description,
-                step_type=step_type,
-                id=id,
-                branch_name=step["branch-name"],
-            )
-        elif step_type == StepType.CHECKOUT:
-            if step.get("branch-name") is None and step.get("commit-hash") is None:
-                raise ValueError(
-                    'Provide either "branch-name" or "commit-hash" in checkout step.'
-                )
-
-            return repo_smith.steps.checkout_step.CheckoutStep(
-                name=name,
-                description=description,
-                step_type=step_type,
-                id=id,
-                branch_name=step.get("branch-name"),
-                commit_hash=step.get("commit-hash"),
-            )
-        elif step_type == StepType.MERGE:
-            if step.get("branch-name") is None:
-                raise ValueError('Provide either "branch-name" in merge step.')
-
-            return repo_smith.steps.merge_step.MergeStep(
-                name=name,
-                description=description,
-                step_type=step_type,
-                id=id,
-                branch_name=step.get("branch-name"),
-                no_fast_forward=step.get("no-ff", False),
-                squash=step.get("squash", False),
-            )
-        elif step_type == StepType.REMOTE:
-            if "remote-url" not in step:
-                raise ValueError('Missing "remote-url" field in remote step.')
-
-            if "remote-name" not in step:
-                raise ValueError('Missing "remote-name" field in remote step.')
-
-            return repo_smith.steps.remote_step.RemoteStep(
-                name=name,
-                description=description,
-                id=id,
-                step_type=step_type,
-                remote_name=step["remote-name"],
-                remote_url=step["remote-url"],
-            )
-        elif step_type == StepType.FETCH:
-            if "remote-name" not in step:
-                raise ValueError('Missing "remote-name" field in fetch step.')
-
-            return repo_smith.steps.fetch_step.FetchStep(
-                name=name,
-                description=description,
-                id=id,
-                step_type=step_type,
-                remote_name=step["remote-name"],
-            )
-        elif step_type in {
-            StepType.NEW_FILE,
-            StepType.APPEND_FILE,
-            StepType.EDIT_FILE,
-            StepType.DELETE_FILE,
-        }:
-            if "filename" not in step:
-                raise ValueError('Missing "filename" field in file step.')
-
-            if step["filename"] is None or step["filename"].strip() == "":
-                raise ValueError('Empty "filename" field in file step.')
-
-            filename = step["filename"]
-            contents = step.get("contents", "") or ""
-
-            match step_type:
-                case StepType.NEW_FILE:
-                    return repo_smith.steps.file_step.NewFileStep(
-                        name=name,
-                        description=description,
-                        step_type=step_type,
-                        id=id,
-                        filename=filename,
-                        contents=contents,
-                    )
-                case StepType.EDIT_FILE:
-                    return repo_smith.steps.file_step.EditFileStep(
-                        name=name,
-                        description=description,
-                        step_type=step_type,
-                        id=id,
-                        filename=filename,
-                        contents=contents,
-                    )
-                case StepType.DELETE_FILE:
-                    return repo_smith.steps.file_step.DeleteFileStep(
-                        name=name,
-                        description=description,
-                        step_type=step_type,
-                        id=id,
-                        filename=filename,
-                        contents=contents,
-                    )
-                case StepType.APPEND_FILE:
-                    return repo_smith.steps.file_step.AppendFileStep(
-                        name=name,
-                        description=description,
-                        step_type=step_type,
-                        id=id,
-                        filename=filename,
-                        contents=contents,
-                    )
-        else:
-            raise ValueError('Improper "type" field in spec.')
-
 
 def initialize_repo(spec_path: str) -> RepoInitializer:
     if not os.path.isfile(spec_path):

src/repo_smith/steps/add_step.py

@@ -1,14 +1,37 @@
-from dataclasses import dataclass
-from typing import List
+from dataclasses import dataclass, field
+from typing import Any, List, Optional, Self, Type
 
 from git import Repo
-
 from repo_smith.steps.step import Step
+from repo_smith.steps.step_type import StepType
 
 
 @dataclass
 class AddStep(Step):
     files: List[str]
 
+    step_type: StepType = field(init=False, default=StepType.ADD)
+
     def execute(self, repo: Repo) -> None:
         repo.index.add(self.files)
+
+    @classmethod
+    def parse(
+        cls: Type[Self],
+        name: Optional[str],
+        description: Optional[str],
+        id: Optional[str],
+        step: Any,
+    ) -> Self:
+        if "files" not in step:
+            raise ValueError('Missing "files" field in add step.')
+
+        if step["files"] is None or step["files"] == []:
+            raise ValueError('Empty "files" list in add step.')
+
+        return cls(
+            name=name,
+            description=description,
+            id=id,
+            files=step["files"],
+        )

src/repo_smith/steps/bash_step.py

@@ -1,15 +1,38 @@
 import subprocess
-
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+from typing import Any, Optional, Self, Type
 
 from git import Repo
-
 from repo_smith.steps.step import Step
+from repo_smith.steps.step_type import StepType
 
 
 @dataclass
 class BashStep(Step):
     body: str
 
-    def execute(self, repo: Repo) -> None:  # type: ignore
+    step_type: StepType = field(init=False, default=StepType.BASH)
+
+    def execute(self, repo: Repo) -> None:
         subprocess.check_call(self.body.strip(), shell=True, cwd=repo.working_dir)
+
+    @classmethod
+    def parse(
+        cls: Type[Self],
+        name: Optional[str],
+        description: Optional[str],
+        id: Optional[str],
+        step: Any,
+    ) -> Self:
+        if "runs" not in step:
+            raise ValueError('Missing "runs" field in bash step.')
+
+        if step["runs"] is None or step["runs"].strip() == "":
+            raise ValueError('Empty "runs" field in bash step.')
+
+        return cls(
+            name=name,
+            description=description,
+            id=id,
+            body=step["runs"],
+        )