Add swimlane

automon/integrations/swimlaneWrapper/api/v2.py

@@ -61,6 +61,24 @@ class Application(object):
     api = f'{Api.api}/app'
     light = f'{api}/light'
 
+    @classmethod
+    def by_id(cls, appId: str):
+        return f'{cls.api}/{appId}'
+
+    @classmethod
+    def export(cls, appId: str):
+        return f'{cls.api}/{appId}/export'
+
+
+class Logging(object):
+    api = f'{Api.api}/logging'
+    job = f'{api}/job'
+    recent = f'{api}/recent'
+
+    @classmethod
+    def by_id(cls, jobId: str):
+        return f'{cls.job}/{jobId}'
+
 
 class Record(object):
 

automon/integrations/swimlaneWrapper/client.py

@@ -76,6 +76,36 @@ async def login_token(self) -> bool:
 
         return response
 
+    async def logging_by_id(self, jobId: str):
+        """finds the recent logs"""
+        url = f'{self.host}/{Logging.by_id(jobId=jobId)}'
+
+        response = await self.requests.get(url=url)
+        logs_job = await self.requests.to_dict()
+
+        return logs_job
+
+    async def logging_recent(self, level: str = 'Debug'):
+        """finds the recent logs"""
+        url = f'{self.host}/{Logging.recent}'
+
+        request_body = {
+            "level": [
+                level
+            ]
+        }
+
+        header = {
+            'Content-Type': 'application/json'
+        }
+
+        # self.requests.session.headers.update(header)
+
+        response = await self.requests.post(url=url, json=request_body, headers=header)
+        logs_recent = await self.requests.to_dict()
+
+        return logs_recent
+
     async def create_auth_token(self):
         """Creates a new access token for the user making the request"""
         url = f'{self.host}/{Auth.create}'
@@ -97,6 +127,28 @@ async def app_list(self):
 
         return self.apps
 
+    async def app_by_id(self, appId: str):
+        url = f'{self.host}/{Application.by_id(appId=appId)}'
+
+        response = await self.requests.get(
+            url=url,
+        )
+
+        app = await self.requests.to_dict()
+
+        return app
+
+    async def app_export(self, appId: str):
+        url = f'{self.host}/{Application.export(appId=appId)}'
+
+        response = await self.requests.get(
+            url=url,
+        )
+
+        app = await self.requests.to_dict()
+
+        return app
+
     @property
     def host(self):
         return self.config.host

automon/integrations/swimlaneWrapper/tests/test_rest_app.py

@@ -1,5 +1,6 @@
 import unittest
 import asyncio
+import random
 
 from automon.integrations.swimlaneWrapper.client import SwimlaneClientRest
 
@@ -10,9 +11,25 @@ class MyTestCase(unittest.TestCase):
     def test_login(self):
         if asyncio.run(client.is_ready()):
             if asyncio.run(client.login()):
-                self.assertTrue(asyncio.run(
+                app_list = asyncio.run(
                     client.app_list())
-                )
+
+                self.assertTrue(app_list)
+
+                app = random.choice(app_list)
+                appId = app['id']
+
+                app_export = asyncio.run(
+                    client.app_export(appId))
+
+                self.assertTrue(app_export)
+
+                app_get = asyncio.run(
+                    client.app_by_id(appId=appId))
+
+                self.assertTrue(app_get)
+
+        pass
 
 
 if __name__ == '__main__':

automon/integrations/swimlaneWrapper/tests/test_rest_logging.py

@@ -0,0 +1,27 @@
+import unittest
+import asyncio
+import random
+
+from automon.integrations.swimlaneWrapper.client import SwimlaneClientRest
+
+client = SwimlaneClientRest()
+
+
+class MyTestCase(unittest.TestCase):
+    def test_login(self):
+        if asyncio.run(client.is_ready()):
+            if asyncio.run(client.login_token()):
+                logs = asyncio.run(
+                    client.logging_recent())
+
+                jobId = '6602057d6158e302aff869b2'
+
+                logs_job = asyncio.run(
+                    client.logging_by_id(jobId=jobId))
+
+
+        pass
+
+
+if __name__ == '__main__':
+    unittest.main()