Tests mostly complete; need to add "action required" tests

.gitignore

@@ -2,6 +2,7 @@
 shinylive/
 shinyapps/*
 # shinyapps/
+.DS_store
 
 .vscode/
 

pyproject.toml

@@ -9,16 +9,20 @@ readme = "README.md"
 
 dependencies = [
   'shiny',
+  'shinylive',
+  'gitpython',
+  'paramiko',
   'pydantic < 2.0',
 ]
 
 [project.optional-dependencies]
 tests = [
   'fastapi',
+  'pytest',
   'loguru',
   'pyfetch_mimic == 2024.06.14',
   'httpx',
-  'shinylive',
+  'pytest',
   'shinyswatch',
 ]
 

src/shinylive_deploy/app.py

@@ -3,13 +3,13 @@
 from pathlib import Path
 
 from pydantic import SecretStr
-from shinylive_deploy.data import create_config_file
 from shinylive_deploy.models import LocalShinyDeploy, ServerShinyDeploy, toml
 
 
 def main():
     config_file = Path.cwd() / "shinylive_deploy.toml"
     if not config_file.exists():
+        from shinylive_deploy.data import create_config_file
         create_config_file()
         return
 
@@ -22,9 +22,11 @@ def main():
         shinylive_.deploy()
 
 
-def _parse_arguments() -> tuple[str, bool]:
+def _parse_arguments(test_argvs: list = None) -> tuple[str, bool]:
+    if test_argvs:
+        sys.argv = test_argvs
     if len(sys.argv) < 2 or sys.argv[1] not in ("test", "beta", "prod", "local"):
-            raise ValueError("\nERROR: One of the following arguments is required -> [ local | test | beta | prod ]\n")
+        raise ValueError("\nERROR: One of the following arguments is required -> [ local | test | beta | prod ]\n")
     deploy_mode = sys.argv[1]
     try:
         rollback = sys.argv[2]
@@ -36,7 +38,7 @@ def _parse_arguments() -> tuple[str, bool]:
     return deploy_mode, rollback
 
 
-def _initialize_configuration(deploy_mode: str):
+def _initialize_configuration(deploy_mode: str) -> LocalShinyDeploy | ServerShinyDeploy:
     if deploy_mode in ("test", "beta", "prod"):
         config = toml["deploy"]["server"]
         return ServerShinyDeploy(

src/shinylive_deploy/data.py

@@ -1,13 +1,11 @@
 from pathlib import Path
 
-
-def create_config_file():
-    text = """
+toml_text = """
 [general]
 app_name = "app1"
 
 [development]
-directory = "src_dev/app1"
+directory = "src"
 
 [deploy.gitbranch]
 prod = "main"
@@ -18,7 +16,7 @@ def create_config_file():
 
 [deploy.local]
 directory = "src_test_webserver/shinyapps/"
-base_url = "localhost:8000/apps"
+base_url = "http://localhost:8000/apps"
 
 [deploy.server]
 host = "127.0.0.1"
@@ -28,6 +26,8 @@ def create_config_file():
 base_url = "http://localhost:5000"
 """
 
-    with open(Path.cwd() / "shinylive_deploy.toml", "w") as f:
-        f.write(text)
-
+
+def create_config_file():
+    filepath = Path.cwd() / "shinylive_deploy.toml"
+    with open(filepath, "w") as f:
+        f.write(toml_text)

src/shinylive_deploy/models/base.py

@@ -6,9 +6,12 @@
 import git
 import tomllib
 
-repo = git.Repo()
+filepath = Path.cwd() / "shinylive_deploy.toml"
+if not filepath.exists():
+    from shinylive_deploy.data import create_config_file
+    create_config_file()
 
-with open("shiny_deploy.toml", "rb") as f:
+with open("shinylive_deploy.toml", "rb") as f:
     toml = tomllib.load(f)
 
 deploy_local = toml["deploy"]["local"]
@@ -18,10 +21,9 @@
 gitbranch: dict = toml["deploy"].get("gitbranch", {})
 
 
-
 @dataclass
 class ShinyDeploy:
-    base_url: str
+    base_url: str = None
     app_name: str = toml["general"]["app_name"]
     dir_deployment: str = None
     dir_development: str = development.get("directory", "src")
@@ -43,16 +45,18 @@ def _message(self):
             "\n##################################"
         )
 
-    def _check_requirements(self):
-        # if len(sys.argv) < 2 or sys.argv[1] not in ("test", "beta", "prod", "local"):
-        #     raise ValueError("\nERROR: One of the following arguments is required -> [ local | test | beta | prod ]\n")
-        # self.mode = sys.argv[1]
+    def _check_git_requirements(self):
+        repo = git.Repo()
         if self.mode == "prod" and str(repo.active_branch) != self.prod_branch:
             raise DeployException(f"Missing Requirement: `prod` deployments can only be executed from the `{self.prod_branch}` branch")
         elif self.mode == "beta" and str(repo.active_branch) != self.beta_branch:
             raise DeployException(f"Missing Requirement: `beta` deployments can only be executed from the `{self.beta_branch}` branch")
 
     def _compile(self):
+        staging_dir = Path.cwd() / "staging"
+        if not staging_dir.exists():
+            staging_dir.mkdir()
+            with open(".gitkeep", "w") as f: f.write("")
         cmd = f"shinylive export {Path(self.dir_development)} {Path(self.dir_staging) / self.deploy_name}"
         print(f"\nExport Command: {cmd}")
         subprocess.run(cmd, shell=True, check=True)

src/shinylive_deploy/models/local.py

@@ -7,11 +7,11 @@
 
 class LocalShinyDeploy(ShinyDeploy):
     def deploy(self):
-        self._check_requirements()
+        self._check_git_requirements()
         self._message()
         self._compile()
 
-        has_backup = self.__manage_backup()
+        has_backup = self._manage_backup()
         if has_backup is None:
             return
 
@@ -20,18 +20,19 @@ def deploy(self):
 
         print(
             "\nCOMPLETE:"
-            f"\n- Application `{self.app_name}` compiled and deployed locally as `{self.deploy_name}`"
+            f"\n- `{self.app_name}` compiled and deployed locally as `{self.deploy_name}`"
             f"\n- Available at {self.base_url}/{self.deploy_name}"
-            f"\n- Backup available at {self.base_url}/{self.deploy_name}-backup" if has_backup is True else ""
         )
+        if has_backup:
+            print(f"\n- Backup available at {self.base_url}/{self.deploy_name}-backup")
 
     def rollback(self):
-        self._check_requirements()
+        self._check_git_requirements()
         existing_deploy_dir = Path(self.dir_deployment) / self.deploy_name
-        if not self.__deployed_dir_exists():
+        if not self._deployed_dir_exists():
             print("\n>>> WARNING <<<: Backback STOPPED. No app directory exists to rollback from.\n")
             return
-        if not self.__backup_dir_exists():
+        if not self._backup_dir_exists():
             print("\n>>> WARNING <<<: Backback STOPPED. No backup directory exists for rollback.\n")
             return
         subprocess.run(f"rm -r {existing_deploy_dir}", **subprocess_config)
@@ -45,24 +46,24 @@ def rollback(self):
             f"\n- Available at {self.base_url}/{self.deploy_name}"
         )
 
-    def __deployed_dir_exists(self):
+    def _deployed_dir_exists(self):
         result = subprocess.run(f"ls {self.dir_deployment}", **subprocess_config)
         directories = result.stdout.split("\n")
         if Path(self.deploy_name).name in directories:
             return True
         return False
 
-    def __backup_dir_exists(self):
+    def _backup_dir_exists(self):
         result = subprocess.run(f"ls {self.dir_deployment}", **subprocess_config)
         directories = result.stdout.split("\n")
         if Path(f"{self.deploy_name}-backup").name in directories:
             return True
         return False
 
-    def __manage_backup(self):
+    def _manage_backup(self):
         existing_deploy_dir = Path(self.dir_deployment) / self.deploy_name
-        if self.__deployed_dir_exists():
-            if self.__backup_dir_exists():
+        if self._deployed_dir_exists():
+            if self._backup_dir_exists():
                 print("\n>>> WARNING <<<: Deployment STOPPED. Backup directory already exists. Delete backup directory, or rollback before redeploying.\n")
                 return None
             subprocess.run(f"mv {existing_deploy_dir} {existing_deploy_dir}-backup", shell=True)

src/shinylive_deploy/models/server.py

@@ -1,12 +1,11 @@
 import subprocess
-import time
 from dataclasses import dataclass
 from pathlib import Path, PurePosixPath
 
 from paramiko import AutoAddPolicy, SFTPClient, SSHClient
 from pydantic import SecretStr
 
-from .base import ShinyDeploy
+from .base import ShinyDeploy, DeployException
 
 subprocess_config = {"capture_output": True, "text": True, "shell": True, "check": True}
 
@@ -25,38 +24,39 @@ def base_ssh_cmd(self):
     def deploy(self, testing: bool = False):
         if not all([self.host, self.user, self.password]):
             raise ValueError("For ServerShinyDeploy, all of the following are required: host, user, password")
-        self._check_requirements()
+        self._check_git_requirements()
         self._message()
         self._compile()
 
         with SSHClient() as ssh:
-            ssh = self.__ssh_connection(ssh)
+            ssh = self._ssh_connection(ssh)
             sftp = ssh.open_sftp()
 
-            has_backup = self.__manage_backup(sftp)
+            has_backup = self._manage_backup(sftp)
             if has_backup is None:
                 return
 
-            self.__push_app(sftp, testing)
+            self._push_app(sftp, testing)
 
         print(
             "\nCOMPLETE:"
             f"\n- `{self.app_name}` compiled and deployed to webserver as `{self.deploy_name}`"
             f"\n- App available at {self.base_url}/{self.deploy_name}"
-            f"\n- Backup available at {self.base_url}/{self.deploy_name}-backup" if has_backup is True else ""
         )
+        if has_backup is True:
+            print(f"\n- Backup available at {self.base_url}/{self.deploy_name}-backup")
 
     def rollback(self):
-        self._check_requirements()
+        self._check_git_requirements()
         deployment_dir = PurePosixPath(self.dir_deployment) / self.deploy_name
 
         with SSHClient() as ssh:
-            ssh = self.__ssh_connection(ssh)
+            ssh = self._ssh_connection(ssh)
             sftp = ssh.open_sftp()
-            if not self.__deployed_dir_exists(sftp):
+            if not self._deployed_dir_exists(sftp):
                 print("\n>>> WARNING <<<: Backback STOPPED. No app directory exists to rollback from.\n")
                 return
-            if not self.__backup_dir_exists(sftp):
+            if not self._backup_dir_exists(sftp):
                 print("\n>>> WARNING <<<: Backback STOPPED. No backup directory exists for rollback.\n")
                 return
 
@@ -71,37 +71,43 @@ def rollback(self):
             f"\n- Available at {self.base_url}/{self.deploy_name}"
         )
 
-    def __ssh_connection(self, client: SSHClient) -> SSHClient:
+    def _ssh_connection(self, client: SSHClient) -> SSHClient:
         client.set_missing_host_key_policy(AutoAddPolicy())
         client.connect(
-            hostname=self.host, port=self.port, username=self.user, password=self.password.get_secret_value()
+            hostname=self.host, port=self.port, username=self.user, password=self.password.get_secret_value(), look_for_keys=False
         )
         return client
 
-    def __deployed_dir_exists(self, sftp: SFTPClient):
+    def _deployed_dir_exists(self, sftp: SFTPClient):
         directories = sftp.listdir(str(self.dir_deployment))
         if self.deploy_name in directories:
             return True
         return False
 
-    def __backup_dir_exists(self, sftp: SFTPClient):
+    def _backup_dir_exists(self, sftp: SFTPClient):
         directories = sftp.listdir(str(self.dir_deployment))
         if f"{self.deploy_name}-backup" in directories:
             return True
         return False
 
-    def __manage_backup(self, sftp: SFTPClient):
+    def _confirm_depoy_dir_exists(self, sftp: SFTPClient):
+        directories = sftp.listdir()
+        if str(self.dir_deployment) not in directories:
+            raise DeployException(f"ACTION REQUIRED`{self.dir_deployment}` not found in the ssh target for user `{self.user}`. Create this directory, owned by this user, then try again.")
+    def _manage_backup(self, sftp: SFTPClient):
+        self._confirm_depoy_dir_exists(sftp)
+
         deployment_filepath = PurePosixPath(self.dir_deployment) / self.deploy_name
         print(deployment_filepath)
-        if self.__deployed_dir_exists(sftp):
-            if self.__backup_dir_exists(sftp):
+        if self._deployed_dir_exists(sftp):
+            if self._backup_dir_exists(sftp):
                 print("\n>>> WARNING <<<: Deployment STOPPED. Backup directory already exists. Delete backup directory, or rollback before redeploying.\n")
                 return None
             sftp.rename(str(deployment_filepath), f"{deployment_filepath}-backup")
             return True
         return False
 
-    def __push_app(self, sftp: SFTPClient, testing: bool = False):
+    def _push_app(self, sftp: SFTPClient, testing: bool = False):
         staging_filepath = Path(self.dir_staging) / self.deploy_name
 
         sftp.mkdir(str(PurePosixPath(self.dir_deployment) / self.deploy_name))

entrypoint.sh → testing/entrypoint.sh

@@ -1,5 +1,4 @@
 #!/bin/bash
-cat /etc/ssh/ssh_config
 
 # Start the first process
 echo "Starting SSH and python webserver..."

testing/src/app.py

@@ -0,0 +1,19 @@
+from module import function_to_import
+from shiny import App, Inputs, Outputs, Session, ui
+
+app_ui = ui.page_fluid(
+    ui.page_sidebar(
+        ui.sidebar(
+            ui.markdown("v2024.06.29b")
+        )
+    ),
+    ui.input_action_button("tests_btn", "Run tests app 1"),
+    title="App1"
+)
+
+
+def server(input: Inputs, output: Outputs, session: Session):
+    function_to_import()
+
+
+app = App(app_ui, server)

testing/src/module/__init__.py

@@ -0,0 +1,2 @@
+def function_to_import():
+    print("yay")