Updated the gitignore to ignore uploaded filed

.gitignore

@@ -23,4 +23,5 @@ safe/
 *.sql
 
 # migrations
-**/migrations/*_initial.py
+**/migrations/*_initial.py
+media/**/*

Pipfile

@@ -21,6 +21,7 @@ django-bootstrap5 = "*"
 django-tables2 = "*"
 requests = "*"
 djangorestframework = "*"
+watchdog = "*"
 
 [dev-packages]
 pycodestyle = "*"

embark/dashboard/views.py

@@ -123,6 +123,7 @@ def show_log(request, analysis_id):
     except FileNotFoundError:
         return HttpResponseServerError(content="File is not yet available")
 
+
 @require_http_methods(["GET"])
 @login_required(login_url='/' + settings.LOGIN_URL)
 def show_logviewer(request, analysis_id):
@@ -141,6 +142,6 @@ def show_logviewer(request, analysis_id):
     logger.debug("Taking file at %s and render it", log_file_path_)
     try:
         return render(request, 'dashboard/logViewer.html', {'analysis_id': analysis_id, 'username': request.user.username})
-    
+
     except FileNotFoundError:
         return HttpResponseServerError(content="File is not yet available")

embark/embark/logviewer.py

@@ -5,7 +5,6 @@
 import json
 
 from pathlib import Path
-import sys
 from watchdog.observers import Observer
 from watchdog.events import FileSystemEventHandler
 from channels.db import database_sync_to_async
@@ -17,6 +16,14 @@
 
 
 class LogConsumer(AsyncWebsocketConsumer):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+        self.analysis_id = None
+        self.line_cache = None
+        self.file_view = None
+        self.observer = None
+
     def load_file_content(self):
         logger.info(
             'Getting file content for analysis id "%s"; view: %s',
@@ -26,15 +33,15 @@ def load_file_content(self):
 
         num_lines = self.line_cache.num_lines()
         limit = min(num_lines, self.file_view.limit)
-    
+
         # Ensure you can't read negative lines
         offset = max(self.file_view.offset, 0)
         # Ensure you can't read lines bigger than the last line
         offset = min(num_lines - limit, offset)
-    
+
         # Fix the offset, in case the user tried to read invalid lines
         self.file_view.offset = offset
-        
+
         content = self.line_cache.read_lines(offset, offset + limit - 1)
         self.file_view.content = base64.b64encode(content).decode("ascii")
         self.file_view.num_lines = num_lines
@@ -53,19 +60,19 @@ async def connect(self):
         firmware = await self.get_firmware(self.analysis_id)
 
         log_file_path_ = f"{Path(firmware.path_to_logs).parent}/emba_run.log"
-        
+
         if not os.path.isfile(log_file_path_):
             await self.send_message({"error": "The log file does not exist, yet."})
             await self.close()
-        
+
         self.line_cache = LineCache(log_file_path_)
 
         self.file_view = FileView()
 
         this = self
 
         class ModifyEventHandler(FileSystemEventHandler):
-            def on_modified(self, event):
+            def on_modified(self, _event):
                 asyncio.run(this.update_lines())
 
         event_handler = ModifyEventHandler()
@@ -94,9 +101,9 @@ async def receive(self, text_data: str = "", bytes_data=None) -> None:
                 self.file_view = FileView(**data["file_view"])
                 await self.send_file_content()
             else:
-                raise Exception("Unknown action")
-        except Exception as e:
-            logger.error(e)
+                raise NotImplementedError("Unknown action")
+        except Exception as exception:
+            logger.error(exception)
             await self.send_message({"error": "Unknown error"})
 
     async def disconnect(self, code):
@@ -123,32 +130,32 @@ def __init__(self, offset=0, limit=30, content="", num_lines=0) -> None:
         self.num_lines = num_lines
 
 
-refresh_lines = 10  # This will define how many of the last lines will be refreshed
+REFRESH_LINES = 10  # This will define how many of the last lines will be refreshed
 
 
 class LineCache:
     def __init__(self, filepath: str) -> None:
         self.line_beginnings = [0]
         self.line_endings = [0]
+        # Intentionally not using with to save resources
+        # because we don't have to open the file as often
+        # pylint: disable-next=consider-using-with
         self.filehandle = open(filepath, "rb")
         self.refresh()
 
     def refresh(self) -> None:
-        refresh_from_line = min(
-            len(self.line_beginnings), refresh_lines
-        )  # Make sure to not go out of bounds
+        # Make sure to not go out of bounds
+        refresh_from_line = min(len(self.line_beginnings), REFRESH_LINES)
         refresh_from_byte = self.line_beginnings[-refresh_from_line]
-        self.line_beginnings = self.line_beginnings[
-            :-refresh_from_line
-        ]  # Remove the line beginnings to be refreshed
-        self.line_endings = self.line_endings[
-            :-refresh_from_line
-        ]  # Remove the line endings to be refreshed
+
+        # Remove the line beginnings and endings to be refreshed
+        self.line_beginnings = self.line_beginnings[:-refresh_from_line]
+        self.line_endings = self.line_endings[:-refresh_from_line]
 
         logger.debug(
-            f"Start refreshing line cache from line {refresh_from_line} (start counting from end of the file)"
+            "Start refreshing line cache from line %s (start counting from end of the file)", refresh_from_line
         )
-        logger.debug(f"Start refreshing line cache from byte {refresh_from_byte}")
+        logger.debug("Start refreshing line cache from byte %s", refresh_from_byte)
 
         self.filehandle.seek(refresh_from_byte)
 
@@ -160,28 +167,28 @@ def refresh(self) -> None:
 
             if len(line) > 0 and line[-1:] == b'\n':
                 line_ending = line_ending - 1
-            
+
             self.line_beginnings.append(line_beginning)
             self.line_endings.append(line_ending)
 
             if len(line) == 0 or line[-1:] != b'\n':
                 break
-                            
+
     def num_lines(self) -> int:
         return len(self.line_beginnings)
 
     def read_lines(self, first_line: int, last_line: int) -> bytes:
         num_lines = self.num_lines()
 
         if first_line > last_line:
-            raise Exception("The first line cannot be bigger than the last line")
-        
+            raise IndexError("The first line cannot be bigger than the last line")
+
         if first_line < 0:
             raise IndexError("The first line cannot be below 0")
 
         if last_line >= num_lines:
             raise IndexError("The first line cannot be equal or above the number of lines")
-        
+
         first_byte = self.line_beginnings[num_lines - last_line - 1]
         last_byte = self.line_endings[num_lines - first_line - 1]
         self.filehandle.seek(first_byte)

embark/embark/tests/test_linecache.py

@@ -2,12 +2,13 @@
 
 from embark.logviewer import LineCache
 
+
 class TestLineCache(unittest.TestCase):
 
     def test_default(self):
         line_cache = LineCache('./test/logviewer/line_cache_test1.log')
 
-        for i in range(0,2):
+        for _ in range(0, 2):
             self.assertEqual(12, line_cache.num_lines(), 'Incorrect number of lines.')
             self.assertEqual(len(line_cache.line_endings), len(line_cache.line_beginnings), 'The number of line beginnings and line endings do not match.')
             self.assertEqual([0, 7, 11, 23, 31, 41, 50, 54, 58, 72, 86, 104], line_cache.line_beginnings, 'The line beginning cache is not valid.')
@@ -19,7 +20,7 @@ def test_default(self):
     def test_no_newline_end(self):
         line_cache = LineCache('./test/logviewer/line_cache_test_no_newline.log')
 
-        for i in range(0,2):
+        for _ in range(0, 2):
             self.assertEqual(11, line_cache.num_lines(), 'Incorrect number of lines.')
             self.assertEqual(len(line_cache.line_endings), len(line_cache.line_beginnings), 'The number of line beginnings and line endings do not match.')
             self.assertEqual([0, 7, 11, 23, 31, 41, 50, 54, 58, 72, 86], line_cache.line_beginnings, 'The line beginning cache is not valid.')
@@ -31,7 +32,7 @@ def test_no_newline_end(self):
     def test_empty(self):
         line_cache = LineCache('./test/logviewer/line_cache_test_empty.log')
 
-        for i in range(0,2):
+        for _ in range(0, 2):
             self.assertEqual(1, line_cache.num_lines(), 'Incorrect number of lines.')
             self.assertEqual(len(line_cache.line_endings), len(line_cache.line_beginnings), 'The number of line beginnings and line endings do not match.')
             self.assertEqual([0], line_cache.line_beginnings, 'The line beginning cache is not valid.')
@@ -41,4 +42,4 @@ def test_empty(self):
 
 
 if __name__ == '__main__':
-    unittest.main()
+    unittest.main()