home · contact · privacy
Enable server to alternatively output response ctx as JSON, for debugging and testing...
authorChristian Heller <c.heller@plomlompom.de>
Sat, 15 Jun 2024 08:47:11 +0000 (10:47 +0200)
committerChristian Heller <c.heller@plomlompom.de>
Sat, 15 Jun 2024 08:47:11 +0000 (10:47 +0200)
plomtask/days.py
plomtask/db.py
plomtask/http.py
plomtask/processes.py
plomtask/versioned_attributes.py
templates/day.html
tests/days.py
tests/utils.py

index afe4a01be6f509a8b624da7c45650500a96805e2..a924bbfeadd2bd895d2f21ac4b7487305fbe771f 100644 (file)
@@ -23,6 +23,13 @@ class Day(BaseModel[str]):
     def __lt__(self, other: Day) -> bool:
         return self.date < other.date
 
+    @property
+    def as_dict(self) -> dict[str, object]:
+        """Return self as (json.dumps-coompatible) dict."""
+        d = super().as_dict
+        d['todos'] = [t.as_dict for t in self.todos]
+        return d
+
     @classmethod
     def from_table_row(cls, db_conn: DatabaseConnection, row: Row | list[Any]
                        ) -> Day:
index 99998a6ab29f760ba0d62f90395739dad4b521ff..1cecc16f6985b555f25757ad6e9f65724311a287 100644 (file)
@@ -271,6 +271,23 @@ class BaseModel(Generic[BaseModelId]):
         assert isinstance(other.id_, int)
         return self.id_ < other.id_
 
+    @property
+    def as_dict(self) -> dict[str, object]:
+        """Return self as (json.dumps-coompatible) dict."""
+        d: dict[str, object] = {'id': self.id_}
+        for k in self.to_save:
+            attr = getattr(self, k)
+            if hasattr(attr, 'as_dict'):
+                d[k] = attr.as_dict
+            d[k] = attr
+        for k in self.to_save_versioned:
+            attr = getattr(self, k)
+            d[k] = attr.as_dict
+        for r in self.to_save_relations:
+            attr_name = r[2]
+            d[attr_name] = [x.as_dict for x in getattr(self, attr_name)]
+        return d
+
     # cache management
     # (we primarily use the cache to ensure we work on the same object in
     # memory no matter where and how we retrieve it, e.g. we don't want
index 26c8b719fec481e4106d66073a85844a1d16491a..d602f07b9baf3c18df19fa88435928d2a33845d6 100644 (file)
@@ -1,11 +1,12 @@
 """Web server stuff."""
 from __future__ import annotations
 from dataclasses import dataclass
-from typing import Any, Callable, Mapping
+from typing import Any, Callable
 from base64 import b64encode, b64decode
 from http.server import BaseHTTPRequestHandler
 from http.server import HTTPServer
 from urllib.parse import urlparse, parse_qs
+from json import dumps as json_dumps
 from os.path import split as path_split
 from jinja2 import Environment as JinjaEnv, FileSystemLoader as JinjaFSLoader
 from plomtask.dating import date_in_n_days
@@ -16,6 +17,7 @@ from plomtask.db import DatabaseConnection, DatabaseFile
 from plomtask.processes import Process, ProcessStep, ProcessStepsNode
 from plomtask.conditions import Condition
 from plomtask.todos import Todo
+from plomtask.db import BaseModel
 
 TEMPLATES_DIR = 'templates'
 
@@ -27,7 +29,37 @@ class TaskServer(HTTPServer):
                  *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
         self.db = db_file
-        self.jinja = JinjaEnv(loader=JinjaFSLoader(TEMPLATES_DIR))
+        self.headers: list[tuple[str, str]] = []
+        self._render_mode = 'html'
+        self._jinja = JinjaEnv(loader=JinjaFSLoader(TEMPLATES_DIR))
+
+    def set_json_mode(self) -> None:
+        """Make server send JSON instead of HTML responses."""
+        self._render_mode = 'json'
+        self.headers += [('Content-Type', 'application/json')]
+
+    @staticmethod
+    def ctx_to_json(ctx: dict[str, object]) -> str:
+        """Render ctx into JSON string."""
+        def walk_ctx(node: object) -> Any:
+            if isinstance(node, BaseModel):
+                return node.as_dict
+            if isinstance(node, (list, tuple)):
+                return [walk_ctx(x) for x in node]
+            if isinstance(node, HandledException):
+                return str(node)
+            return node
+        for k, v in ctx.items():
+            ctx[k] = walk_ctx(v)
+        return json_dumps(ctx)
+
+    def render(self, ctx: dict[str, object], tmpl_name: str = '') -> str:
+        """Render ctx according to self._render_mode.."""
+        tmpl_name = f'{tmpl_name}.{self._render_mode}'
+        if 'html' == self._render_mode:
+            template = self._jinja.get_template(tmpl_name)
+            return template.render(ctx)
+        return self.__class__.ctx_to_json(ctx)
 
 
 class InputsParser:
@@ -106,16 +138,18 @@ class TaskHandler(BaseHTTPRequestHandler):
     _form_data: InputsParser
     _params: InputsParser
 
-    def _send_html(self,
+    def _send_page(self,
+                   ctx: dict[str, Any],
                    tmpl_name: str,
-                   ctx: Mapping[str, object],
-                   code: int = 200) -> None:
+                   code: int = 200
+                   ) -> None:
         """Send HTML as proper HTTP response."""
-        tmpl = self.server.jinja.get_template(tmpl_name)
-        html = tmpl.render(ctx)
+        body = self.server.render(ctx, tmpl_name)
         self.send_response(code)
+        for header_tuple in self.server.headers:
+            self.send_header(*header_tuple)
         self.end_headers()
-        self.wfile.write(bytes(html, 'utf-8'))
+        self.wfile.write(bytes(body, 'utf-8'))
 
     @staticmethod
     def _request_wrapper(http_method: str, not_found_msg: str
@@ -148,7 +182,7 @@ class TaskHandler(BaseHTTPRequestHandler):
                         assert hasattr(cls, 'empty_cache')
                         cls.empty_cache()
                     ctx = {'msg': error}
-                    self._send_html('msg.html', ctx, error.http_code)
+                    self._send_page(ctx, 'msg', error.http_code)
                 finally:
                     self.conn.close()
             return wrapper
@@ -158,11 +192,11 @@ class TaskHandler(BaseHTTPRequestHandler):
     def do_GET(self, handler: Callable[[], str | dict[str, object]]
                ) -> str | None:
         """Render page with result of handler, or redirect if result is str."""
-        tmpl_name = f'{self._site}.html'
-        ctx_or_redir = handler()
-        if isinstance(ctx_or_redir, str):
-            return ctx_or_redir
-        self._send_html(tmpl_name, ctx_or_redir)
+        tmpl_name = f'{self._site}'
+        ctx_or_redir_target = handler()
+        if isinstance(ctx_or_redir_target, str):
+            return ctx_or_redir_target
+        self._send_page(ctx_or_redir_target, tmpl_name)
         return None
 
     @_request_wrapper('POST', 'Unknown POST target')
index 06ee4ba9b9c2a3011019b03b1b0e21633fce780c..6df8eaf9132d7652e447931d59a2cc6a39d3043b 100644 (file)
@@ -46,6 +46,14 @@ class Process(BaseModel[int], ConditionsRelations):
         self.calendarize = calendarize
         self.n_owners: int | None = None  # only set by from_table_row
 
+    @property
+    def as_dict(self) -> dict[str, object]:
+        """Return self as (json.dumps-coompatible) dict."""
+        d = super().as_dict
+        d['explicit_steps'] = [s.as_dict for s in self.explicit_steps]
+        d['suppressed_steps'] = [s.as_dict for s in self.suppressed_steps]
+        return d
+
     @classmethod
     def from_table_row(cls, db_conn: DatabaseConnection,
                        row: Row | list[Any]) -> Process:
@@ -218,7 +226,7 @@ class ProcessStep(BaseModel[int]):
         self.parent_step_id = parent_step_id
 
     def save(self, db_conn: DatabaseConnection) -> None:
-        """Remove from DB, and owner's .explicit_steps."""
+        """Update into DB/cache, and owner's .explicit_steps."""
         super().save(db_conn)
         owner = Process.by_id(db_conn, self.owner_id)
         if self not in owner.explicit_steps:
index cbd1c8e348a9230b10176d55b4b6a490fe11ff33..b7e54e287f9844fe9cc910dcf53ac15cc1a8421e 100644 (file)
@@ -25,6 +25,14 @@ class VersionedAttribute:
                     history_tuples)
         return hash(hashable)
 
+    @property
+    def as_dict(self) -> dict[str, object]:
+        """Return self as (json.dumps-coompatible) dict."""
+        d = {'parent_process_id': self.parent.id_,
+             'table_name': self.table_name,
+             'history': self.history}
+        return d
+
     @property
     def _newest_timestamp(self) -> str:
         """Return most recent timestamp."""
index f980cd1f5ad93d271e8103c981c8cdf49523e8fe..acc9aaa7524f8ffed1dc98f34fddb02ce6278541 100644 (file)
@@ -142,10 +142,12 @@ comment:
 add: <input type="text" name="new_todo" list="processes">
 </p>
 <p>
+make new todos
 <select name="make_type">
-<option value="full">make new todos with children</option>
-<option value="empty"{% if make_type == "empty" %}selected {% endif %}>make new todos without children</option>
+<option value="full">with</option>
+<option value="empty"{% if make_type == "empty" %}selected {% endif %}>without</option>
 </select>
+descendants (i.e. adopt where possible, otherwise create anew)
 </p>
 
 <table>
index 286f75815ef51e74ade9e15e96bdb85dc4218a4b..901667f4c6e0276a2800bf4b21b15b03a17be2fb 100644 (file)
@@ -1,6 +1,7 @@
 """Test Days module."""
 from unittest import TestCase
 from datetime import datetime
+from json import loads as json_loads
 from tests.utils import TestCaseWithDB, TestCaseWithServer
 from plomtask.dating import date_in_n_days
 from plomtask.days import Day
@@ -105,6 +106,23 @@ class TestsWithDB(TestCaseWithDB):
 class TestsWithServer(TestCaseWithServer):
     """Tests against our HTTP server/handler (and database)."""
 
+    def test_get_json(self) -> None:
+        """Test /day for JSON response."""
+        self.conn.request('GET', '/day?date=2024-01-01')
+        response = self.conn.getresponse()
+        self.assertEqual(response.status, 200)
+        expected = {'day': {'id': '2024-01-01',
+                            'comment': '',
+                            'todos': []},
+                    'top_nodes': [],
+                    'make_type': '',
+                    'enablers_for': {},
+                    'disablers_for': {},
+                    'conditions_present': [],
+                    'processes': []}
+        retrieved = json_loads(response.read().decode())
+        self.assertEqual(expected, retrieved)
+
     def test_do_GET(self) -> None:
         """Test /day and /calendar response codes, and / redirect."""
         self.check_get('/day', 200)
index a9a4e80418a54f288281999b964ce8364474c177..15a53ae0ddc0b78835b5baacea15f43d3a81cba0 100644 (file)
@@ -217,6 +217,7 @@ class TestCaseWithServer(TestCaseWithDB):
         self.server_thread.start()
         self.conn = HTTPConnection(str(self.httpd.server_address[0]),
                                    self.httpd.server_address[1])
+        self.httpd.set_json_mode()
 
     def tearDown(self) -> None:
         self.httpd.shutdown()