Merge pull request #18 from Pilifer/master

Helper script to delete log files

Author: Glenn Jones

pythonanywhere/api.py

@@ -214,3 +214,34 @@ def delete_log(self, log_type, index=0):
                     response_text=response.text,
                 )
             )
+
+    def get_log_info(self):
+        url = get_api_endpoint().format(username=getpass.getuser(),
+                                        flavor="files") + "tree/?path=/var/log/"
+        response = call_api(url, "get")
+        if not response.ok:
+            raise Exception(
+                "GET log files info via API failed, got {response}:{response_text}".format(
+                    response=response,
+                    response_text=response.text,
+                )
+            )
+        file_list = response.json()
+        log_types = ["access", "error", "server"]
+        logs = {"access": [], "error": [], "server": []}
+        log_prefix = "/var/log/{domain}.".format(domain=self.domain)
+        for file_name in file_list:
+            if type(file_name) == str and file_name.startswith(log_prefix):
+                log = file_name[len(log_prefix):].split(".")
+                if log[0] in log_types:
+                    log_type = log[0]
+                    if log[-1] == "log":
+                        log_index = 0
+                    elif log[-1] == "1":
+                        log_index = 1
+                    elif log[-1] == "gz":
+                        log_index = int(log[-2])
+                    else:
+                        continue
+                    logs[log_type].append(log_index)
+        return logs

scripts/pa_autoconfigure_django.py

@@ -12,7 +12,7 @@
 
 Options:
   --domain=<domain>         Domain name, eg www.mydomain.com   [default: your-username.pythonanywhere.com]
-  --python=<python-version> Python version, eg "2.7"    [default: 3.6]
+  --python=<python-version> Python version, eg "2.7"    [default: 3.5]
   --nuke                    *Irrevocably* delete any existing web app config on this domain. Irrevocably.
 """
 

scripts/pa_delete_webapp_logs.py

@@ -0,0 +1,50 @@
+#!/usr/bin/python3.5
+"""Deletes webapp logs.
+
+- gets list of logs via api
+- deletes logs via api
+
+Usage:
+  pa_delete_webapp_logs.py [--domain=<domain>] [--log_type=<log_type>] [--log_index=<log_index>]
+
+Options:
+  --domain=<domain>         Domain name, eg www.mydomain.com   [default: your-username.pythonanywhere.com]
+  --log_type=<log_type>     Log type, could be access, error, server or all   [default: all]
+  --log_index=<log_index>   Log index, 0 for current log, 1-9 for one of archive logs or all [default: all]
+"""
+
+from docopt import docopt
+import getpass
+import os
+
+from pythonanywhere.api import Webapp
+from pythonanywhere.snakesay import snakesay
+
+
+def main(domain, log_type, log_index):
+    if domain == 'your-username.pythonanywhere.com':
+        username = getpass.getuser().lower()
+        pa_domain = os.environ.get('PYTHONANYWHERE_DOMAIN', 'pythonanywhere.com')
+        domain = '{username}.{pa_domain}'.format(username=username, pa_domain=pa_domain)
+
+    webapp = Webapp(domain)
+    log_types = ["access", "error", "server"]
+    logs = webapp.get_log_info()
+    if log_type == "all" and log_index == "all":
+        for key in log_types:
+            for log in logs[key]:
+                webapp.delete_log(key, log)
+    elif log_type == "all":
+        for key in log_types:
+            webapp.delete_log(key, int(log_index))
+    elif log_index == "all":
+        for i in logs[log_type]:
+            webapp.delete_log(log_type, int(i))
+    else:
+        webapp.delete_log(log_type, int(log_index))
+    print(snakesay('All Done!'))
+
+
+if __name__ == '__main__':
+    arguments = docopt(__doc__)
+    main(arguments['--domain'], arguments['--log_type'], arguments['--log_index'])

tests/test_api.py

@@ -356,7 +356,7 @@ def test_delete_old_access_log(self, api_responses, api_token):
         assert post.request.body is None
         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):
+    def test_raises_if_delete_does_not_20x(self, api_responses, api_token):
         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")
@@ -366,3 +366,38 @@ def test_raises_if_post_does_not_20x(self, api_responses, api_token):
 
         assert "DELETE log file via API failed" in str(e.value)
         assert "nope" in str(e.value)
+
+
+class TestGetWebappLogs:
+
+    def test_get_list_of_logs(self, api_responses, api_token):
+        expected_url = get_api_endpoint().format(
+            username=getpass.getuser(), flavor="files") + "tree/?path=/var/log/"
+        api_responses.add(responses.GET, expected_url, status=200,
+                          body="['/var/log/blah','/var/log/mydomain.com.access.log',\
+                          '/var/log/mydomain.com.access.log.1',\
+                          '/var/log/mydomain.com.access.log.2.gz',\
+                          '/var/log/mydomain.com.error.log',\
+                          '/var/log/mydomain.com.error.log.1',\
+                          '/var/log/mydomain.com.error.log.2.gz',\
+                          '/var/log/mydomain.com.server.log',\
+                          '/var/log/mydomain.com.server.log.1',\
+                          '/var/log/mydomain.com.server.log.2.gz']")
+
+        logs = Webapp("mydomain.com").get_log_info()
+
+        post = api_responses.calls[0]
+        assert post.request.url == expected_url
+        assert post.request.headers['Authorization'] == 'Token {api_token}'.format(api_token=api_token)
+        assert logs == {"access": [0, 1, 2], "error": [0, 1, 2], "server": [0, 1, 2]}
+
+    def test_raises_if_get_does_not_20x(self, api_responses, api_token):
+        expected_url = get_api_endpoint().format(
+            username=getpass.getuser(), flavor="files") + "tree/?path=/var/log/"
+        api_responses.add(responses.GET, expected_url, status=404, body="nope")
+
+        with pytest.raises(Exception) as e:
+            Webapp("mydomain.com").get_log_info()
+
+        assert "GET log files info via API failed" in str(e.value)
+        assert "nope" in str(e.value)