Implement get_log_file method returning dictionary of lists of indices of each webapp log type based on list of files gathered using API.

Author: filiplajszczak

pythonanywhere/api.py

@@ -214,3 +214,38 @@ 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,
+                )
+            )
+        if response.text[0] == "[" and response.text[-1] == "]":
+            file_list = eval(response.text)
+        else:
+            raise Exception(
+                "GET log files info via API failed, got {text} instead of list of files".format(text=response.text))
+        log_types = ["access", "error", "server"]
+        logs = {"access": [], "error": [], "server": []}
+        log_prefix = "/var/log/{domain}.".format(domain=self.domain)
+        for file in file_list:
+            if type(file) == str and file.startswith(log_prefix):
+                log = file[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

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,48 @@ 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)
+
+    def test_raises_if_response_text_not_list(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="print('Behold malicious code')")
+
+        with pytest.raises(Exception) as e:
+            Webapp("mydomain.com").get_log_info()
+
+        assert "GET log files info via API failed, got print('Behold malicious code')" in str(e.value)