Table views by name. Started drafting 'sources' view

tcsfw/client_api.py

@@ -127,7 +127,7 @@ def api_get(self, request: APIRequest, pretty=False)  -> str:
         context = RequestContext(request, self)
         path = request.path
         if path.startswith("table/"):
-            text = TableView.get_print(self.registry.system, path[6:], parameters=request.parameters)
+            text = TableView.get_print(self.registry, path[6:], parameters=request.parameters)
             return text
         if path == "all":
             request.get_connections = False

tcsfw/client_tool.py

@@ -65,6 +65,7 @@ def run(self):
 
         # Subcommand: views
         show_parser = subparsers.add_parser("show", help="Show view")
+        show_parser.add_argument("view", help="Name of the view")
         show_parser.add_argument("--url", "-u", help="Server URL")
         show_parser.add_argument("--api-key", help="API key for server (avoiding providing by command line)")
 
@@ -286,11 +287,12 @@ def upload_file_data(self, url: str, temp_file: BinaryIO):
 
     def run_show(self, args: argparse.Namespace):
         """Textual views"""
+        view = args.view
         url = args.url.strip()
         if not url:
             raise ValueError("Missing server URL")
         api_url = self.resolve_api_url(url)
-        table_url = f"{api_url}/table/xxx"
+        table_url = f"{api_url}/table/{view}"
         headers = {}
         if self.auth_token:
             headers["X-Authorization"] = self.auth_token

tcsfw/entity_database.py

@@ -3,7 +3,7 @@
 import logging
 from typing import Any, Iterable, Optional, Dict, List
 from tcsfw.event_interface import EventInterface
-from tcsfw.traffic import Event
+from tcsfw.traffic import Event, EvidenceSource
 
 
 class EntityDatabase:
@@ -40,6 +40,10 @@ def put_event(self, event: Event):
     def clear_database(self):
         """Clear the database, from the disk"""
 
+    def get_souces(self) -> List[EvidenceSource]:
+        """Get evidence sources"""
+        return []
+
 
 class InMemoryDatabase(EntityDatabase):
     """Store and retrieve events, later entities, etc."""
@@ -80,3 +84,9 @@ def put_event(self, event: Event):
         self.trail.append(event)
         source = event.evidence.source
         self.trail_filter.setdefault(source.label, True)
+
+    def get_souces(self) -> List[EvidenceSource]:
+        sources = {}
+        for e in self.trail:
+            sources.setdefault(e.evidence.source, e)
+        return list(sources.keys())

tcsfw/registry.py

@@ -1,7 +1,7 @@
 """Event registry backed by database"""
 
 import logging
-from typing import Optional, Dict, Self, Any, Set
+from typing import List, Optional, Dict, Self, Any, Set
 
 from tcsfw.entity import Entity
 from tcsfw.entity_database import EntityDatabase, InMemoryDatabase

tcsfw/text_tables.py

@@ -6,6 +6,7 @@
 from tcsfw.basics import Status
 from tcsfw.entity import Entity
 from tcsfw.model import Host, IoTSystem, NetworkNode, Service
+from tcsfw.registry import Registry
 
 
 class BaseTable:
@@ -125,22 +126,48 @@ def print(self, stream: TextIO):
         self.print_rows(rows, stream)
 
 
+class SourceTable(BaseTable):
+    """Source table"""
+    def __init__(self, registry: Registry, screen_size: Tuple[int, int]):
+        super().__init__([
+            ("Source", 20),
+            ("Reference", 40),
+            ("Status", 10),
+            ("Age", 10),
+        ], screen_size)
+        self.registry = registry
+
+    def print(self, stream: TextIO):
+        rows = [[h[0] for h in self.columns]]
+        sources = self.registry.database.get_souces()
+        for src in sources:
+            rows.append([src.name, src.base_ref, "", ""])
+
+        self.print_rows(rows, stream)
+
+
 class TableView:
     """View of one or more tables"""
     def __init__(self, tables: List[BaseTable]) -> None:
         self.tables = tables
 
     @classmethod
-    def get_print(cls, model: IoTSystem, _name: str, parameters: Dict[str, str]) -> str:
+    def get_print(cls, registry: Registry, name: str, parameters: Dict[str, str]) -> str:
         """Get printout by name"""
         screen = parameters.get("screen")
         if screen:
             screen_size = tuple([int(x) for x in screen.split("x")])
             assert len(screen_size) == 2
         else:
             screen_size = (80, 50)
-        screen_size = screen_size[0], int(screen_size[1] / 2)
-        view = TableView([HostTable(model, screen_size), ConnectionTable(model, screen_size)])
+        model = registry.get_system()
+        if name == "system":
+            screen_size = screen_size[0], int(screen_size[1] / 2)
+            view = TableView([HostTable(model, screen_size), ConnectionTable(model, screen_size)])
+        elif name == "sources":
+            view = TableView([SourceTable(registry, screen_size)])
+        else:
+            raise ValueError(f"Unnown view '{name}'")
         buf = StringIO()
         view.print(buf)
         return buf.getvalue()