web app creation.

Author: filiplajszczak

pythonanywhere_core/webapp.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import os
+import getpass
+from pathlib import Path
+from textwrap import dedent
+
+from snakesay import snakesay
+
+from pythonanywhere_core.base import call_api, get_api_endpoint, PYTHON_VERSIONS
+from pythonanywhere_core.exceptions import SanityException, PythonAnywhereApiException
+
+
+class Webapp:
+    def __init__(self, domain: str) -> None:
+        self.domain = domain
+
+    def __eq__(self, other: Webapp) -> bool:
+        return self.domain == other.domain
+
+    def sanity_checks(self, nuke: bool) -> None:
+        print(snakesay("Running API sanity checks"))
+        token = os.environ.get("API_TOKEN")
+        if not token:
+            raise SanityException(
+                dedent(
+                    """
+                Could not find your API token.
+                You may need to create it on the Accounts page?
+                You will also need to close this console and open a new one once you've done that.
+                """
+                )
+            )
+
+        if nuke:
+            return
+
+        endpoint = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps")
+        url = f"{endpoint}{self.domain}/"
+        response = call_api(url, "get")
+        if response.status_code == 200:
+            raise SanityException(
+                f"You already have a webapp for {self.domain}.\n\nUse the --nuke option if you want to replace it."
+            )
+
+    def create(self, python_version: str, virtualenv_path: Path, project_path: Path, nuke: bool) -> None:
+        print(snakesay("Creating web app via API"))
+        base_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps")
+        domain_url = f"{base_url}{self.domain}/"
+        if nuke:
+            call_api(domain_url, "delete")
+        patch_url = domain_url
+        response = call_api(
+            base_url, "post", data={"domain_name": self.domain, "python_version": PYTHON_VERSIONS[python_version]}
+        )
+        if not response.ok or response.json().get("status") == "ERROR":
+            raise PythonAnywhereApiException(
+                f"POST to create webapp via API failed, got {response}:{response.text}"
+            )
+        response = call_api(
+            patch_url, "patch", data={"virtualenv_path": virtualenv_path, "source_directory": project_path}
+        )
+        if not response.ok:
+            raise PythonAnywhereApiException(
+                "PATCH to set virtualenv path and source directory via API failed,"
+                "got {response}:{response_text}".format(response=response, response_text=response.text)
+            )

tests/conftest.py

@@ -1,7 +1,15 @@
 import os
+import shutil
+from getpass import getuser
+from pathlib import Path
 
 import pytest
 import responses
+import tempfile
+
+
+def _get_temp_dir():
+    return Path(tempfile.mkdtemp())
 
 
 @pytest.fixture
@@ -21,4 +29,71 @@ def api_token():
     if old_token is None:
         del os.environ["API_TOKEN"]
     else:
-        os.environ["API_TOKEN"] = old_token
+        os.environ["API_TOKEN"] = old_token
+
+
+@pytest.fixture(scope="session")
+def local_pip_cache(request):
+    previous_cache = request.config.cache.get("pythonanywhere/pip-cache", None)
+    if previous_cache:
+        return Path(previous_cache)
+    else:
+        new_cache = _get_temp_dir()
+        request.config.cache.set("pythonanywhere/pip-cache", str(new_cache))
+        return new_cache
+
+
+@pytest.fixture
+def fake_home(local_pip_cache):
+    tempdir = _get_temp_dir()
+    cache_dir = tempdir / ".cache"
+    cache_dir.mkdir()
+    (cache_dir / "pip").symlink_to(local_pip_cache)
+
+    old_home = os.environ["HOME"]
+    old_home_contents = set(Path(old_home).iterdir())
+
+    os.environ["HOME"] = str(tempdir)
+    yield tempdir
+    os.environ["HOME"] = old_home
+    shutil.rmtree(str(tempdir), ignore_errors=True)
+
+    new_stuff = set(Path(old_home).iterdir()) - old_home_contents
+    if new_stuff:
+        raise Exception(f"home mocking failed somewehere: {new_stuff}, {tempdir}")
+
+
+@pytest.fixture
+def virtualenvs_folder():
+    actual_virtualenvs = Path(f"/home/{getuser()}/.virtualenvs")
+    if actual_virtualenvs.is_dir():
+        old_virtualenvs = set(Path(actual_virtualenvs).iterdir())
+    else:
+        old_virtualenvs = {}
+
+    tempdir = _get_temp_dir()
+    old_workon = os.environ.get("WORKON_HOME")
+    os.environ["WORKON_HOME"] = str(tempdir)
+    yield tempdir
+    if old_workon:
+        os.environ["WORKON_HOME"] = old_workon
+    else:
+        del os.environ["WORKON_HOME"]
+    shutil.rmtree(str(tempdir), ignore_errors=True)
+
+    if actual_virtualenvs.is_dir():
+        new_envs = set(actual_virtualenvs.iterdir()) - set(old_virtualenvs)
+        if new_envs:
+            raise Exception(f"virtualenvs path mocking failed somewehere: {new_envs}, {tempdir}")
+
+
+@pytest.fixture(scope="function")
+def no_api_token():
+    if "API_TOKEN" not in os.environ:
+        yield
+
+    else:
+        old_token = os.environ["API_TOKEN"]
+        del os.environ["API_TOKEN"]
+        yield
+        os.environ["API_TOKEN"] = old_token

tests/test_webapp.py

@@ -0,0 +1,215 @@
+import getpass
+import json
+
+import pytest
+import responses
+from urllib.parse import urlencode
+
+from pythonanywhere_core.base import get_api_endpoint, PYTHON_VERSIONS
+from pythonanywhere_core.exceptions import SanityException, PythonAnywhereApiException
+from pythonanywhere_core.webapp import Webapp
+
+
+def test_init():
+    app = Webapp("www.my-domain.com")
+    assert app.domain == "www.my-domain.com"
+
+
+def test_compare_equal():
+    assert Webapp("www.my-domain.com") == Webapp("www.my-domain.com")
+
+
+def test_compare_not_equal():
+    assert Webapp("www.my-domain.com") != Webapp("www.other-domain.com")
+
+
+@pytest.fixture
+def domain():
+    return "www.domain.com"
+
+
+@pytest.fixture
+def base_url():
+    return get_api_endpoint().format(username=getpass.getuser(), flavor='webapps')
+
+
+@pytest.fixture
+def domain_url(base_url, domain):
+    return f"{base_url}{domain}/"
+
+
+@pytest.fixture
+def webapp(domain):
+    return Webapp(domain)
+
+
+def test_does_not_complain_if_api_token_exists(api_token, api_responses, domain_url, webapp):
+    api_responses.add(responses.GET, domain_url, status=404)
+    webapp.sanity_checks(nuke=False)  # should not raise
+
+
+def test_raises_if_no_api_token_exists(api_responses, no_api_token, webapp):
+    with pytest.raises(SanityException) as e:
+        webapp.sanity_checks(nuke=False)
+    assert "Could not find your API token" in str(e.value)
+
+
+def test_raises_if_webapp_already_exists(api_token, api_responses, domain, domain_url, webapp):
+    api_responses.add(
+        responses.GET,
+        domain_url,
+        status=200,
+        body=json.dumps({"id": 1, "domain_name": domain}),
+    )
+
+    with pytest.raises(SanityException) as e:
+        webapp.sanity_checks(nuke=False)
+
+    assert f"You already have a webapp for {domain}" in str(e.value)
+    assert "nuke" in str(e.value)
+
+
+def test_does_not_raise_if_no_webapp(api_token, api_responses, domain_url, webapp):
+    api_responses.add(responses.GET, domain_url, status=404)
+    webapp.sanity_checks(nuke=False)  # should not raise
+
+
+def test_nuke_option_overrides_all_but_token_check(
+    api_token, api_responses, domain, fake_home, virtualenvs_folder, webapp
+):
+    (fake_home / domain).mkdir()
+    (virtualenvs_folder / domain).mkdir()
+
+    webapp.sanity_checks(nuke=True)  # should not raise
+
+
+def test_does_post_to_create_webapp(api_responses, api_token, base_url, domain, domain_url, webapp):
+    api_responses.add(
+        responses.POST,
+        base_url,
+        status=201,
+        body=json.dumps({"status": "OK"}),
+    )
+    api_responses.add(responses.PATCH, domain_url, status=200)
+
+    webapp.create(
+        "3.10", "/virtualenv/path", "/project/path", nuke=False
+    )
+
+    post = api_responses.calls[0]
+    assert post.request.url == base_url
+    assert post.request.body == urlencode(
+        {"domain_name": domain, "python_version": PYTHON_VERSIONS["3.10"]}
+    )
+    assert post.request.headers["Authorization"] == f"Token {api_token}"
+
+
+def test_does_patch_to_update_virtualenv_path_and_source_directory(
+    api_responses, api_token, base_url, domain_url, webapp
+):
+    api_responses.add(
+        responses.POST,
+        base_url,
+        status=201,
+        body=json.dumps({"status": "OK"}),
+    )
+    api_responses.add(responses.PATCH, domain_url, status=200)
+
+    webapp.create(
+        "3.10", "/virtualenv/path", "/project/path", nuke=False
+    )
+
+    patch = api_responses.calls[1]
+    assert patch.request.url == domain_url
+    assert patch.request.body == urlencode(
+        {"virtualenv_path": "/virtualenv/path", "source_directory": "/project/path"}
+    )
+    assert patch.request.headers["Authorization"] == f"Token {api_token}"
+
+
+def test_raises_if_post_does_not_20x(api_responses, api_token, base_url, webapp):
+    api_responses.add(
+        responses.POST, base_url, status=500, body="an error"
+    )
+
+    with pytest.raises(PythonAnywhereApiException) as e:
+        webapp.create(
+            "3.10", "/virtualenv/path", "/project/path", nuke=False
+        )
+
+    assert "POST to create webapp via API failed" in str(e.value)
+    assert "an error" in str(e.value)
+
+
+def test_raises_if_post_returns_a_200_with_status_error(api_responses, api_token, base_url, webapp):
+    api_responses.add(
+        responses.POST,
+        base_url,
+        status=200,
+        body=json.dumps(
+            {
+                "status": "ERROR",
+                "error_type": "bad",
+                "error_message": "bad things happened",
+            }
+        ),
+    )
+
+    with pytest.raises(PythonAnywhereApiException) as e:
+        webapp.create(
+            "3.10", "/virtualenv/path", "/project/path", nuke=False
+        )
+
+    assert "POST to create webapp via API failed" in str(e.value)
+    assert "bad things happened" in str(e.value)
+
+
+def test_raises_if_patch_does_not_20x(api_responses, api_token, base_url, domain_url, webapp):
+    api_responses.add(
+        responses.POST,
+        base_url,
+        status=201,
+        body=json.dumps({"status": "OK"}),
+    )
+    api_responses.add(
+        responses.PATCH,
+        domain_url,
+        status=400,
+        json={"message": "an error"},
+    )
+
+    with pytest.raises(PythonAnywhereApiException) as e:
+        webapp.create(
+            "3.7", "/virtualenv/path", "/project/path", nuke=False
+        )
+
+    assert (
+        "PATCH to set virtualenv path and source directory via API failed"
+        in str(e.value)
+    )
+    assert "an error" in str(e.value)
+
+
+def test_does_delete_first_for_nuke_call(api_responses, api_token, base_url, domain_url, webapp):
+    api_responses.add(responses.DELETE, domain_url, status=200)
+    api_responses.add(
+        responses.POST, base_url, status=201, body=json.dumps({"status": "OK"})
+    )
+    api_responses.add(responses.PATCH, domain_url, status=200)
+
+    webapp.create("3.10", "/virtualenv/path", "/project/path", nuke=True)
+
+    delete = api_responses.calls[0]
+    assert delete.request.method == "DELETE"
+    assert delete.request.url == domain_url
+    assert delete.request.headers["Authorization"] == f"Token {api_token}"
+
+
+def test_ignores_404_from_delete_call_when_nuking(api_responses, api_token, base_url, domain_url, webapp):
+    api_responses.add(responses.DELETE, domain_url, status=404)
+    api_responses.add(
+        responses.POST, base_url, status=201, body=json.dumps({"status": "OK"})
+    )
+    api_responses.add(responses.PATCH, domain_url, status=200)
+
+    webapp.create("3.10", "/virtualenv/path", "/project/path", nuke=True)