test: add tests for deploy and cli commands

src/writer/command_line.py

@@ -50,7 +50,7 @@ def edit(path, port, host, enable_remote_edit, enable_server_setup):
         enable_remote_edit=enable_remote_edit, enable_server_setup=enable_server_setup)
 
 @main.command()
-@click.argument('path', type=click.Path(exists=True, file_okay=False, dir_okay=True, resolve_path=True))
+@click.argument('path', type=click.Path(exists=False, file_okay=False, dir_okay=True, resolve_path=True))
 @click.option('--template', help="The template to use when creating a new app.")
 def create(path, template):
     """Create a new app in PATH folder."""

src/writer/deploy.py

@@ -15,7 +15,6 @@
 import requests
 from gitignore_parser import parse_gitignore
 
-WRITER_DEPLOY_URL = os.getenv("WRITER_DEPLOY_URL", "https://api.writer.com/v1/deployment/apps")
 
 @click.group()
 def cloud():
@@ -32,20 +31,26 @@ def cloud():
     hide_input=True, help="Writer API key"
 )
 @click.option('--env', '-e', multiple=True, default=[], help="Environment to deploy the app to")
+@click.option('--force', '-f', default=False, is_flag=True, help="Ignores warnings and overwrites the app")
 @click.option('--verbose', '-v', default=False, is_flag=True, help="Enable verbose mode")
 @click.argument('path')
-def deploy(path, api_key, env, verbose):
-    check_app(api_key)
+def deploy(path, api_key, env, verbose, force):
     """Deploy the app from PATH folder."""
 
+    deploy_url = os.getenv("WRITER_DEPLOY_URL", "https://api.writer.com/v1/deployment/apps")
+    sleep_interval = int(os.getenv("WRITER_DEPLOY_SLEEP_INTERVAL", '5'))
+
+    if not force:
+        check_app(deploy_url, api_key)
+
     abs_path, is_folder = _get_absolute_app_path(path)
     if not is_folder:
         raise click.ClickException("A path to a folder containing a Writer Framework app is required. For example: writer cloud deploy my_app")
 
     env = _validate_env_vars(env)
     tar = pack_project(abs_path)
     try:
-        upload_package(tar, api_key, env, verbose=verbose)
+        upload_package(deploy_url, tar, api_key, env, verbose=verbose, sleep_interval=sleep_interval)
     except requests.exceptions.HTTPError as e:
         if e.response.status_code == 401:
             unauthorized_error()
@@ -83,7 +88,8 @@ def undeploy(api_key, verbose):
     """Stop the app, app would not be available anymore."""
     try:
         print("Undeploying app")
-        with requests.delete(WRITER_DEPLOY_URL, headers={"Authorization": f"Bearer {api_key}"}) as resp:
+        deploy_url = os.getenv("WRITER_DEPLOY_URL", "https://api.writer.com/v1/deployment/apps")
+        with requests.delete(deploy_url, headers={"Authorization": f"Bearer {api_key}"}) as resp:
             on_error_print_and_raise(resp, verbose=verbose)
             print("App undeployed")
             sys.exit(0)
@@ -105,13 +111,16 @@ def undeploy(api_key, verbose):
 def logs(api_key, verbose):
     """Fetch logs from the deployed app."""
 
+    deploy_url = os.getenv("WRITER_DEPLOY_URL", "https://api.writer.com/v1/deployment/apps")
+    sleep_interval = int(os.getenv("WRITER_DEPLOY_SLEEP_INTERVAL", '5'))
+
     try: 
         build_time = datetime.now(pytz.timezone('UTC')) - timedelta(days=4)
         start_time = build_time
         while True:
             prev_start = start_time
             end_time = datetime.now(pytz.timezone('UTC'))
-            data = get_logs(api_key, {
+            data = get_logs(deploy_url, api_key, {
                 "buildTime": build_time,
                 "startTime": start_time,
                 "endTime": end_time,
@@ -122,12 +131,12 @@ def logs(api_key, verbose):
                 start_time = start_time if start_time > log[0] else log[0]
             if start_time == prev_start:
                 start_time = datetime.now(pytz.timezone('UTC'))
-                time.sleep(5)
+                time.sleep(sleep_interval)
                 continue
             for log in logs:
                 print(log[0], log[1])
             print(start_time)
-            time.sleep(1)
+            time.sleep(sleep_interval)
     except Exception as e:
         print(e)
         sys.exit(1)
@@ -164,17 +173,17 @@ def match(file_path) -> bool: return False
 
     return f
 
-def check_app(token):
-    url = get_app_url(token)
+def check_app(deploy_url, token):
+    url = get_app_url(deploy_url, token)
     if url:
         print("[WARNING] This token was already used to deploy a different app")
         print(f"[WARNING] URL: {url}")
         print("[WARNING] If looking to deploy to a different URL, use a different API key. ")
         if input("[WARNING] Are you sure you want to overwrite? (y/N)").lower() != "y":
             sys.exit(1)
 
-def get_app_url(token):
-    with requests.get(WRITER_DEPLOY_URL, params={"lineLimit": 1}, headers={"Authorization": f"Bearer {token}"}) as resp:
+def get_app_url(deploy_url, token):
+    with requests.get(deploy_url, params={"lineLimit": 1}, headers={"Authorization": f"Bearer {token}"}) as resp:
         try:
             resp.raise_for_status()
         except Exception as e:
@@ -184,8 +193,8 @@ def get_app_url(token):
         data = resp.json()
     return data['status']['url']
 
-def get_logs(token, params, verbose=False):
-    with requests.get(WRITER_DEPLOY_URL, params = params, headers={"Authorization": f"Bearer {token}"}) as resp:
+def get_logs(deploy_url, token, params, verbose=False):
+    with requests.get(deploy_url, params = params, headers={"Authorization": f"Bearer {token}"}) as resp:
         on_error_print_and_raise(resp, verbose=verbose)
         data = resp.json()
 
@@ -198,8 +207,8 @@ def get_logs(token, params, verbose=False):
         logs.sort(key=lambda x: x[0])
         return {"status": data["status"], "logs": logs}
 
-def check_service_status(token, build_id, build_time, start_time, end_time, last_status):
-    data = get_logs(token, {
+def check_service_status(deploy_url, token, build_id, build_time, start_time, end_time, last_status):
+    data = get_logs(deploy_url, token, {
         "buildId": build_id,
         "buildTime": build_time,
         "startTime": start_time,
@@ -225,14 +234,15 @@ def dictFromEnv(env: List[str]) -> dict:
     return env_dict
 
 
-def upload_package(tar, token, env, verbose=False):
+def upload_package(deploy_url, tar, token, env, verbose=False, sleep_interval=5):
     print("Uploading package to deployment server")
     tar.seek(0)
     files = {'file': tar}
     start_time = datetime.now(pytz.timezone('UTC'))
     build_time = start_time
+
     with requests.post(
-        url = WRITER_DEPLOY_URL, 
+        url = deploy_url, 
         headers = {
             "Authorization": f"Bearer {token}",
         },
@@ -248,17 +258,17 @@ def upload_package(tar, token, env, verbose=False):
     url = ""
     while status not in ["COMPLETED", "FAILED"] and datetime.now(pytz.timezone('UTC')) < build_time + timedelta(minutes=5):
         end_time = datetime.now(pytz.timezone('UTC'))
-        status, url = check_service_status(token, build_id, build_time, start_time, end_time, status)
-        time.sleep(5)
+        status, url = check_service_status(deploy_url, token, build_id, build_time, start_time, end_time, status)
+        time.sleep(sleep_interval)
         start_time = end_time
 
     if status == "COMPLETED":
         print("Deployment successful")
         print(f"URL: {url}")
         sys.exit(0)
     else:
-        time.sleep(5)
-        check_service_status(token, build_id, build_time, start_time, datetime.now(pytz.timezone('UTC')), status)
+        time.sleep(sleep_interval)
+        check_service_status(deploy_url, token, build_id, build_time, start_time, datetime.now(pytz.timezone('UTC')), status)
         print("Deployment failed")
         sys.exit(1)
 
@@ -271,7 +281,6 @@ def on_error_print_and_raise(resp, verbose=False):
         raise e
 
 def unauthorized_error():
-    print(f"\n{WRITER_DEPLOY_URL}")
     print("Unauthorized. Please check your API key.")
     sys.exit(1)
 

tests/backend/test_cli.py

@@ -0,0 +1,42 @@
+import os
+
+from click.testing import CliRunner
+from writer.command_line import main
+
+
+def test_version():
+   runner = CliRunner()
+   with runner.isolated_filesystem():
+      result = runner.invoke(main, ['-v'])
+      assert result.exit_code == 0
+      assert 'version' in result.output
+
+def test_create_default():
+    runner = CliRunner()
+    with runner.isolated_filesystem():
+        result = runner.invoke(main, ['create', './my_app'])
+        print(result.output)
+        assert result.exit_code == 0
+        #check if filder exists and if has the right files
+        assert os.path.exists('./my_app')
+        assert os.path.exists('./my_app/ui.json')
+        assert os.path.exists('./my_app/main.py')
+        #load toml and check name and version
+        with open('./my_app/pyproject.toml') as f:
+            content = f.read()
+        assert content.find('name = "writer-framework-default"') != -1
+
+def test_create_specific_template():
+    runner = CliRunner()
+    with runner.isolated_filesystem():
+        result = runner.invoke(main, ['create', './my_app', '--template', 'hello'])
+        print(result.output)
+        assert result.exit_code == 0
+        #check if filder exists and if has the right files
+        assert os.path.exists('./my_app')
+        assert os.path.exists('./my_app/ui.json')
+        assert os.path.exists('./my_app/main.py')
+        #load toml and check name and version
+        with open('./my_app/pyproject.toml') as f:
+            content = f.read()
+        assert content.find('name = "writer-framework-hello"') != -1