Merge pull request #17 from Pilifer/master

 replacing flavor="webapps" default hack with nice {{flavor}}

Author: gpjt

pythonanywhere/api.py

@@ -22,12 +22,14 @@
 class AuthenticationError(Exception):
     pass
 
+
 class NoTokenError(Exception):
     pass
 
-def get_api_endpoint(flavor="webapps"):
+
+def get_api_endpoint():
     domain = os.environ.get('PYTHONANYWHERE_DOMAIN', 'pythonanywhere.com')
-    return 'https://www.{domain}/api/v0/user/{{username}}/{flavor}/'.format(domain=domain, flavor=flavor)
+    return 'https://www.{domain}/api/v0/user/{{username}}/{{flavor}}/'.format(domain=domain)
 
 
 def call_api(url, method, **kwargs):
@@ -84,7 +86,7 @@ def sanity_checks(self, nuke):
         if nuke:
             return
 
-        url = get_api_endpoint().format(username=getpass.getuser()) + self.domain + '/'
+        url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + self.domain + '/'
         response = call_api(url, 'get')
         if response.status_code == 200:
             raise SanityException(
@@ -98,9 +100,9 @@ def sanity_checks(self, nuke):
     def create(self, python_version, virtualenv_path, project_path, nuke):
         print(snakesay('Creating web app via API'))
         if nuke:
-            webapp_url = get_api_endpoint().format(username=getpass.getuser()) + self.domain + '/'
+            webapp_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + self.domain + '/'
             call_api(webapp_url, 'delete')
-        post_url = get_api_endpoint().format(username=getpass.getuser())
+        post_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps")
         patch_url = post_url + self.domain + '/'
         response = call_api(post_url, 'post', data={
             'domain_name': self.domain, 'python_version': PYTHON_VERSIONS[python_version]},
@@ -130,7 +132,7 @@ def create(self, python_version, virtualenv_path, project_path, nuke):
     def add_default_static_files_mappings(self, project_path):
         print(snakesay('Adding static files mappings for /static/ and /media/'))
 
-        url = get_api_endpoint().format(username=getpass.getuser()) + self.domain + '/static_files/'
+        url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + self.domain + '/static_files/'
         call_api(url, 'post', json=dict(
             url='/static/', path=str(Path(project_path) / 'static'),
         ))
@@ -142,7 +144,7 @@ def add_default_static_files_mappings(self, project_path):
 
     def reload(self):
         print(snakesay('Reloading {domain} via API'.format(domain=self.domain)))
-        url = get_api_endpoint().format(username=getpass.getuser()) + self.domain + '/reload/'
+        url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + self.domain + '/reload/'
         response = call_api(url, 'post')
         if not response.ok:
             raise Exception(
@@ -155,7 +157,7 @@ def reload(self):
 
     def set_ssl(self, certificate, private_key):
         print(snakesay('Setting up SSL for {domain} via API'.format(domain=self.domain)))
-        url = get_api_endpoint().format(username=getpass.getuser()) + self.domain + '/ssl/'
+        url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + self.domain + '/ssl/'
         response = call_api(
             url, 'post',
             json={'cert': certificate, 'private_key': private_key}
@@ -170,7 +172,7 @@ def set_ssl(self, certificate, private_key):
 
 
     def get_ssl_info(self):
-        url = get_api_endpoint().format(username=getpass.getuser()) + self.domain + '/ssl/'
+        url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + self.domain + '/ssl/'
         response = call_api(url, 'get')
         if not response.ok:
             raise Exception(
@@ -195,14 +197,14 @@ def delete_log(self, log_type, index=0):
                 'Deleting current {type} log file for {domain} via API'.format(type=log_type, domain=self.domain)))
 
         if index == 1:
-            url = get_api_endpoint("files").format(username=getpass.getuser()) + "path/var/log/{domain}.{type}.log.1/".format(
+            url = get_api_endpoint().format(username=getpass.getuser(), flavor="files") + "path/var/log/{domain}.{type}.log.1/".format(
                 domain=self.domain, type=log_type)
         elif index > 1:
-            url = get_api_endpoint("files").format(
-                username=getpass.getuser()) + "path/var/log/{domain}.{type}.log.{index}.gz/".format(
+            url = get_api_endpoint().format(
+                username=getpass.getuser(), flavor="files") + "path/var/log/{domain}.{type}.log.{index}.gz/".format(
                 domain=self.domain, type=log_type, index=index)
         else:
-            url = get_api_endpoint("files").format(username=getpass.getuser()) + "path/var/log/{domain}.{type}.log/".format(
+            url = get_api_endpoint().format(username=getpass.getuser(), flavor="files") + "path/var/log/{domain}.{type}.log/".format(
                 domain=self.domain, type=log_type)
         response = call_api(url, "delete")
         if not response.ok:

tests/test_api.py

@@ -19,9 +19,9 @@
 class TestGetAPIEndpoint:
 
     def test_gets_domain_from_env_if_set(self, monkeypatch):
-        assert get_api_endpoint() == 'https://www.pythonanywhere.com/api/v0/user/{username}/webapps/'
+        assert get_api_endpoint() == 'https://www.pythonanywhere.com/api/v0/user/{username}/{flavor}/'
         monkeypatch.setenv('PYTHONANYWHERE_DOMAIN', 'foo.com')
-        assert get_api_endpoint() == 'https://www.foo.com/api/v0/user/{username}/webapps/'
+        assert get_api_endpoint() == 'https://www.foo.com/api/v0/user/{username}/{flavor}/'
 
 
 class TestCallAPI:
@@ -67,7 +67,7 @@ def test_compare_not_equal(self):
 
 class TestWebappSanityChecks:
     domain = 'www.domain.com'
-    expected_url = get_api_endpoint().format(username=getpass.getuser()) + domain + '/'
+    expected_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + domain + '/'
 
     def test_does_not_complain_if_api_token_exists(self, api_token, api_responses):
         webapp = Webapp(self.domain)
@@ -115,8 +115,8 @@ def test_nuke_option_overrides_all_but_token_check(
 class TestCreateWebapp:
 
     def test_does_post_to_create_webapp(self, api_responses, api_token):
-        expected_post_url = get_api_endpoint().format(username=getpass.getuser())
-        expected_patch_url = get_api_endpoint().format(username=getpass.getuser()) + 'mydomain.com/'
+        expected_post_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps")
+        expected_patch_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + 'mydomain.com/'
         api_responses.add(responses.POST, expected_post_url, status=201, body=json.dumps({'status': 'OK'}))
         api_responses.add(responses.PATCH, expected_patch_url, status=200)
 
@@ -132,8 +132,8 @@ def test_does_post_to_create_webapp(self, api_responses, api_token):
 
 
     def test_does_patch_to_update_virtualenv_path_and_source_directory(self, api_responses, api_token):
-        expected_post_url = get_api_endpoint().format(username=getpass.getuser())
-        expected_patch_url = get_api_endpoint().format(username=getpass.getuser()) + 'mydomain.com/'
+        expected_post_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps")
+        expected_patch_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + 'mydomain.com/'
         api_responses.add(responses.POST, expected_post_url, status=201, body=json.dumps({'status': 'OK'}))
         api_responses.add(responses.PATCH, expected_patch_url, status=200)
 
@@ -149,7 +149,7 @@ def test_does_patch_to_update_virtualenv_path_and_source_directory(self, api_res
 
 
     def test_raises_if_post_does_not_20x(self, api_responses, api_token):
-        expected_post_url = get_api_endpoint().format(username=getpass.getuser())
+        expected_post_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps")
         api_responses.add(responses.POST, expected_post_url, status=500, body='an error')
 
         with pytest.raises(Exception) as e:
@@ -160,7 +160,7 @@ def test_raises_if_post_does_not_20x(self, api_responses, api_token):
 
 
     def test_raises_if_post_returns_a_200_with_status_error(self, api_responses, api_token):
-        expected_post_url = get_api_endpoint().format(username=getpass.getuser())
+        expected_post_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps")
         api_responses.add(responses.POST, expected_post_url, status=200, body=json.dumps({
             "status": "ERROR", "error_type": "bad", "error_message": "bad things happened"
         }))
@@ -173,8 +173,8 @@ def test_raises_if_post_returns_a_200_with_status_error(self, api_responses, api
 
 
     def test_raises_if_patch_does_not_20x(self, api_responses, api_token):
-        expected_post_url = get_api_endpoint().format(username=getpass.getuser())
-        expected_patch_url = get_api_endpoint().format(username=getpass.getuser()) + 'mydomain.com/'
+        expected_post_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps")
+        expected_patch_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + 'mydomain.com/'
         api_responses.add(responses.POST, expected_post_url, status=201, body=json.dumps({'status': 'OK'}))
         api_responses.add(responses.PATCH, expected_patch_url, status=400, json={'message': 'an error'})
 
@@ -186,8 +186,8 @@ def test_raises_if_patch_does_not_20x(self, api_responses, api_token):
 
 
     def test_does_delete_first_for_nuke_call(self, api_responses, api_token):
-        post_url = get_api_endpoint().format(username=getpass.getuser())
-        webapp_url = get_api_endpoint().format(username=getpass.getuser()) + 'mydomain.com/'
+        post_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps")
+        webapp_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + 'mydomain.com/'
         api_responses.add(responses.DELETE, webapp_url, status=200)
         api_responses.add(responses.POST, post_url, status=201, body=json.dumps({'status': 'OK'}))
         api_responses.add(responses.PATCH, webapp_url, status=200)
@@ -201,8 +201,8 @@ def test_does_delete_first_for_nuke_call(self, api_responses, api_token):
 
 
     def test_ignores_404_from_delete_call_when_nuking(self, api_responses, api_token):
-        post_url = get_api_endpoint().format(username=getpass.getuser())
-        webapp_url = get_api_endpoint().format(username=getpass.getuser()) + 'mydomain.com/'
+        post_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps")
+        webapp_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + 'mydomain.com/'
         api_responses.add(responses.DELETE, webapp_url, status=404)
         api_responses.add(responses.POST, post_url, status=201, body=json.dumps({'status': 'OK'}))
         api_responses.add(responses.PATCH, webapp_url, status=200)
@@ -214,7 +214,7 @@ def test_ignores_404_from_delete_call_when_nuking(self, api_responses, api_token
 class TestAddDefaultStaticFilesMapping:
 
     def test_does_two_posts_to_static_files_endpoint(self, api_token, api_responses):
-        expected_url = get_api_endpoint().format(username=getpass.getuser()) + 'mydomain.com/static_files/'
+        expected_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + 'mydomain.com/static_files/'
         api_responses.add(responses.POST, expected_url, status=201)
         api_responses.add(responses.POST, expected_url, status=201)
 
@@ -240,7 +240,7 @@ def test_does_two_posts_to_static_files_endpoint(self, api_token, api_responses)
 class TestReloadWebapp:
 
     def test_does_post_to_reload_url(self, api_responses, api_token):
-        expected_url = get_api_endpoint().format(username=getpass.getuser()) + 'mydomain.com/reload/'
+        expected_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + 'mydomain.com/reload/'
         api_responses.add(responses.POST, expected_url, status=200)
 
         Webapp('mydomain.com').reload()
@@ -252,7 +252,7 @@ def test_does_post_to_reload_url(self, api_responses, api_token):
 
 
     def test_raises_if_post_does_not_20x(self, api_responses, api_token):
-        expected_url = get_api_endpoint().format(username=getpass.getuser()) + 'mydomain.com/reload/'
+        expected_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + 'mydomain.com/reload/'
         api_responses.add(responses.POST, expected_url, status=404, body='nope')
 
         with pytest.raises(Exception) as e:
@@ -265,7 +265,7 @@ def test_raises_if_post_does_not_20x(self, api_responses, api_token):
 class TestSetWebappSSL:
 
     def test_does_post_to_ssl_url(self, api_responses, api_token):
-        expected_url = get_api_endpoint().format(username=getpass.getuser()) + 'mydomain.com/ssl/'
+        expected_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + 'mydomain.com/ssl/'
         api_responses.add(responses.POST, expected_url, status=200)
         certificate = "certificate data"
         private_key = "private key data"
@@ -281,7 +281,7 @@ def test_does_post_to_ssl_url(self, api_responses, api_token):
 
 
     def test_raises_if_post_does_not_20x(self, api_responses, api_token):
-        expected_url = get_api_endpoint().format(username=getpass.getuser()) + 'mydomain.com/ssl/'
+        expected_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + 'mydomain.com/ssl/'
         api_responses.add(responses.POST, expected_url, status=404, body='nope')
 
         with pytest.raises(Exception) as e:
@@ -294,7 +294,7 @@ def test_raises_if_post_does_not_20x(self, api_responses, api_token):
 class TestGetWebappSSLInfo:
 
     def test_returns_json_from_server_having_parsed_expiry(self, api_responses, api_token):
-        expected_url = get_api_endpoint().format(username=getpass.getuser()) + 'mydomain.com/ssl/'
+        expected_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + 'mydomain.com/ssl/'
         api_responses.add(
             responses.GET, expected_url, status=200,
             body=json.dumps({
@@ -320,7 +320,7 @@ def test_returns_json_from_server_having_parsed_expiry(self, api_responses, api_
 
 
     def test_raises_if_get_does_not_return_200(self, api_responses, api_token):
-        expected_url = get_api_endpoint().format(username=getpass.getuser()) + 'mydomain.com/ssl/'
+        expected_url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + 'mydomain.com/ssl/'
         api_responses.add(responses.GET, expected_url, status=404, body='nope')
 
         with pytest.raises(Exception) as e:
@@ -333,8 +333,8 @@ def test_raises_if_get_does_not_return_200(self, api_responses, api_token):
 class TestDeleteWebappLog:
 
     def test_delete_current_access_log(self, api_responses, api_token):
-        expected_url = get_api_endpoint(flavor="files").format(
-            username=getpass.getuser()) + "path/var/log/mydomain.com.access.log/"
+        expected_url = get_api_endpoint().format(
+            username=getpass.getuser(), flavor="files") + "path/var/log/mydomain.com.access.log/"
         api_responses.add(responses.DELETE, expected_url, status=200)
 
         Webapp("mydomain.com").delete_log(log_type="access")
@@ -345,8 +345,8 @@ def test_delete_current_access_log(self, api_responses, api_token):
         assert post.request.headers['Authorization'] == 'Token {api_token}'.format(api_token=api_token)
 
     def test_delete_old_access_log(self, api_responses, api_token):
-        expected_url = get_api_endpoint(flavor="files").format(
-            username=getpass.getuser()) + "path/var/log/mydomain.com.access.log.1/"
+        expected_url = get_api_endpoint().format(
+            username=getpass.getuser(), flavor="files") + "path/var/log/mydomain.com.access.log.1/"
         api_responses.add(responses.DELETE, expected_url, status=200)
 
         Webapp("mydomain.com").delete_log(log_type="access", index=1)
@@ -357,8 +357,8 @@ def test_delete_old_access_log(self, api_responses, api_token):
         assert post.request.headers['Authorization'] == 'Token {api_token}'.format(api_token=api_token)
 
     def test_raises_if_post_does_not_20x(self, api_responses, api_token):
-        expected_url = get_api_endpoint(flavor="files").format(
-            username=getpass.getuser()) + "path/var/log/mydomain.com.access.log/"
+        expected_url = get_api_endpoint().format(
+            username=getpass.getuser(), flavor="files") + "path/var/log/mydomain.com.access.log/"
         api_responses.add(responses.DELETE, expected_url, status=404, body="nope")
 
         with pytest.raises(Exception) as e: