|
|
@@ -0,0 +1,424 @@
|
|
|
+#!/usr/bin/env python3
|
|
|
+"""Static + logic tests for the task event-log page's client JS.
|
|
|
+
|
|
|
+The app binary cannot be built in every container (statum-mkpstm /
|
|
|
+libstatum are build-time dependencies), so this suite verifies
|
|
|
+src/pages/task_log.html two ways:
|
|
|
+
|
|
|
+1. Statically - the <script> body must be ASCII-only (the page
|
|
|
+ compiler HTML-escapes script bodies, so any non-ASCII literal is
|
|
|
+ re-served as literal entity text like "·" after the
|
|
|
+ browser's one decode pass), must wire the SSE stream through
|
|
|
+ `onmessage` (the server sends id:/data: frames with no `event:`
|
|
|
+ name, so named listeners can never fire), must not listen for an
|
|
|
+ 'append' event name, and its delimiter structure must balance
|
|
|
+ after strings and comments are stripped.
|
|
|
+
|
|
|
+2. Behaviourally - the ingest / render-order / scroll-stick /
|
|
|
+ load-more logic is ported verbatim below as pure python functions
|
|
|
+ and unit-tested: oldest-first chronology, id dedupe, the 48px
|
|
|
+ bottom-stick rule, prepend-mode scroll anchoring, and the
|
|
|
+ newest-first-page reversal with the `more` -> exhausted flag.
|
|
|
+
|
|
|
+Runnable standalone: python3 test/test_task_log_js.py
|
|
|
+"""
|
|
|
+
|
|
|
+import os
|
|
|
+import re
|
|
|
+import unittest
|
|
|
+
|
|
|
+PAGE = os.path.join(os.path.dirname(os.path.abspath(__file__)),
|
|
|
+ os.pardir, "src", "pages", "task_log.html")
|
|
|
+
|
|
|
+
|
|
|
+def script_body():
|
|
|
+ with open(PAGE, encoding="utf-8") as handle:
|
|
|
+ source = handle.read()
|
|
|
+ head, rest = source.split("<script>", 1)
|
|
|
+ body, _ = rest.split("</script>", 1)
|
|
|
+ assert head, "no content before the script block"
|
|
|
+ return body
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+# Static checks
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+
|
|
|
+class ScriptSourceTests(unittest.TestCase):
|
|
|
+ """What the served bytes must look like after compilation."""
|
|
|
+
|
|
|
+ def setUp(self):
|
|
|
+ self.body = script_body()
|
|
|
+
|
|
|
+ def test_script_is_ascii_only(self):
|
|
|
+ """Non-ASCII in a script body comes back as literal entity
|
|
|
+ text ("·") because the compiler escapes script bodies
|
|
|
+ and the browser decodes exactly once. ASCII (with \\uXXXX
|
|
|
+ escapes for the characters we want) is immune."""
|
|
|
+ offenders = [(index, hex(ord(char)))
|
|
|
+ for index, char in enumerate(self.body) if ord(char) > 127]
|
|
|
+ self.assertEqual(offenders, [])
|
|
|
+
|
|
|
+ def test_unicode_escapes_used_for_separators(self):
|
|
|
+ """The separator characters must still render - as escapes,
|
|
|
+ not as raw bytes and not as HTML entities."""
|
|
|
+ for escape in ("\\u00b7", "\\u2026", "\\u2014"):
|
|
|
+ self.assertIn(escape, self.body)
|
|
|
+ self.assertNotIn("·", self.body)
|
|
|
+ self.assertNotIn("—", self.body)
|
|
|
+
|
|
|
+ def test_stream_wired_through_onmessage(self):
|
|
|
+ self.assertRegex(self.body, r"stream\.onmessage\s*=")
|
|
|
+ self.assertNotIn("addEventListener('append'", self.body)
|
|
|
+ self.assertNotIn('addEventListener("append"', self.body)
|
|
|
+
|
|
|
+ def test_heartbeat_ping_is_skipped(self):
|
|
|
+ self.assertRegex(self.body, r"message\.data\s*===\s*'ping'")
|
|
|
+
|
|
|
+ def test_load_more_pages_before_oldest(self):
|
|
|
+ self.assertRegex(self.body, r"/events\?before='\s*\+\s*\(state\.oldest")
|
|
|
+
|
|
|
+ def test_page_is_reversed_before_ingest(self):
|
|
|
+ self.assertIn("slice().reverse().forEach(ingest)", self.body)
|
|
|
+
|
|
|
+ def test_exhausted_set_when_no_more(self):
|
|
|
+ self.assertIn("state.exhausted = true", self.body)
|
|
|
+
|
|
|
+ def test_more_button_sits_above_the_feed(self):
|
|
|
+ # Search the page markup only - the script's comments also
|
|
|
+ # mention [data-log-feed], before the elements do.
|
|
|
+ with open(PAGE, encoding="utf-8") as handle:
|
|
|
+ markup = handle.read().split("</script>", 1)[1]
|
|
|
+ more = markup.index("data-log-more")
|
|
|
+ feed = markup.index("data-log-feed")
|
|
|
+ self.assertLess(more, feed,
|
|
|
+ "the load-older control must precede the feed markup")
|
|
|
+
|
|
|
+ def test_delimiters_balance(self):
|
|
|
+ """A cheap structural stand-in for a JS parser (none is
|
|
|
+ installed here): after stripping comments and string literals
|
|
|
+ every bracket must close."""
|
|
|
+ body = self.body
|
|
|
+ out = []
|
|
|
+ i = 0
|
|
|
+ mode = None # None | 'line' | 'block' | "'" | '"'
|
|
|
+ while i < len(body):
|
|
|
+ char = body[i]
|
|
|
+ nxt = body[i + 1] if i + 1 < len(body) else ""
|
|
|
+ if mode is None:
|
|
|
+ if char == "/" and nxt == "/":
|
|
|
+ mode = "line"
|
|
|
+ i += 2
|
|
|
+ continue
|
|
|
+ if char == "/" and nxt == "*":
|
|
|
+ mode = "block"
|
|
|
+ i += 2
|
|
|
+ continue
|
|
|
+ if char in "'\"":
|
|
|
+ mode = char
|
|
|
+ i += 1
|
|
|
+ continue
|
|
|
+ out.append(char)
|
|
|
+ elif mode == "line":
|
|
|
+ if char == "\n":
|
|
|
+ mode = None
|
|
|
+ out.append("\n")
|
|
|
+ elif mode == "block":
|
|
|
+ if char == "*" and nxt == "/":
|
|
|
+ mode = None
|
|
|
+ i += 2
|
|
|
+ continue
|
|
|
+ else: # inside a string literal
|
|
|
+ if char == "\\":
|
|
|
+ i += 2
|
|
|
+ continue
|
|
|
+ if char == mode:
|
|
|
+ mode = None
|
|
|
+ i += 1
|
|
|
+ self.assertEqual(mode, None, "unterminated comment or string literal")
|
|
|
+ pairs = {")": "(", "]": "[", "}": "{"}
|
|
|
+ stack = []
|
|
|
+ for char in out:
|
|
|
+ if char in "([{":
|
|
|
+ stack.append(char)
|
|
|
+ elif char in pairs:
|
|
|
+ self.assertTrue(stack and stack[-1] == pairs[char],
|
|
|
+ "unbalanced %r at depth %s" % (char, stack))
|
|
|
+ stack.pop()
|
|
|
+ self.assertEqual(stack, [], "unclosed delimiters %s" % stack)
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+# Port of the client logic (keep in step with task_log.html)
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+
|
|
|
+STICK_THRESHOLD = 48 # px that still counts as "reading the tail"
|
|
|
+PAGE_SIZE = 30
|
|
|
+
|
|
|
+
|
|
|
+class Feed(object):
|
|
|
+ """Scroll-state stand-in for [data-log-feed]."""
|
|
|
+
|
|
|
+ def __init__(self, scroll_height=0, client_height=600, scroll_top=0):
|
|
|
+ self.scroll_height = scroll_height
|
|
|
+ self.client_height = client_height
|
|
|
+ self.scroll_top = scroll_top
|
|
|
+
|
|
|
+
|
|
|
+def at_bottom(feed):
|
|
|
+ """PORT: a feed shorter than its viewport counts as at-bottom."""
|
|
|
+ return feed.scroll_height - feed.scroll_top - feed.client_height \
|
|
|
+ <= STICK_THRESHOLD
|
|
|
+
|
|
|
+
|
|
|
+class State(object):
|
|
|
+ """PORT: the module-level `state` object."""
|
|
|
+
|
|
|
+ def __init__(self):
|
|
|
+ self.events = []
|
|
|
+ self.oldest = 0
|
|
|
+ self.exhausted = False
|
|
|
+ self.loading = False
|
|
|
+ self.seeded = False
|
|
|
+
|
|
|
+
|
|
|
+def ingest(state, evt):
|
|
|
+ """PORT: keep state.events oldest-first (ascending id); skip ids
|
|
|
+ already held (SSE replay after a reconnect resends a frame)."""
|
|
|
+ i = 0
|
|
|
+ while i < len(state.events) and state.events[i]["id"] < evt["id"]:
|
|
|
+ i += 1
|
|
|
+ if i < len(state.events) and state.events[i]["id"] == evt["id"]:
|
|
|
+ return
|
|
|
+ state.events.insert(i, evt)
|
|
|
+ if evt["id"] < state.oldest or state.oldest == 0:
|
|
|
+ state.oldest = evt["id"]
|
|
|
+
|
|
|
+
|
|
|
+def render(state, feed, mode=None, height_after=None):
|
|
|
+ """PORT of render(mode): decide the post-render scroll position.
|
|
|
+ height_after stands in for the DOM rebuild's new scrollHeight."""
|
|
|
+ was_at_bottom = at_bottom(feed)
|
|
|
+ height_before = feed.scroll_height
|
|
|
+ scroll_top_before = feed.scroll_top
|
|
|
+ if height_after is not None:
|
|
|
+ feed.scroll_height = height_after
|
|
|
+ if not state.seeded and len(state.events) > 0:
|
|
|
+ state.seeded = True
|
|
|
+ feed.scroll_top = feed.scroll_height # scrollToBottom()
|
|
|
+ elif mode == "prepend":
|
|
|
+ if was_at_bottom:
|
|
|
+ feed.scroll_top = feed.scroll_height
|
|
|
+ else:
|
|
|
+ feed.scroll_top = scroll_top_before + (feed.scroll_height
|
|
|
+ - height_before)
|
|
|
+ elif was_at_bottom:
|
|
|
+ feed.scroll_top = feed.scroll_height
|
|
|
+ else:
|
|
|
+ feed.scroll_top = scroll_top_before
|
|
|
+ return feed.scroll_top
|
|
|
+
|
|
|
+
|
|
|
+def server_page(newest, oldest, more):
|
|
|
+ """The endpoint's wire shape: newest-first within the page."""
|
|
|
+ return {"events": [{"id": i} for i in range(newest, oldest - 1, -1)],
|
|
|
+ "more": more}
|
|
|
+
|
|
|
+
|
|
|
+def load_more(state, feed, page, height_after):
|
|
|
+ """PORT of loadMore's success path: reverse the newest-first page
|
|
|
+ into chronological order, ingest it, set exhausted, render."""
|
|
|
+ rows = page["events"]
|
|
|
+ for evt in reversed(rows):
|
|
|
+ ingest(state, evt)
|
|
|
+ if not page["more"] or len(rows) == 0:
|
|
|
+ state.exhausted = True
|
|
|
+ return render(state, feed, mode="prepend", height_after=height_after)
|
|
|
+
|
|
|
+
|
|
|
+def on_stream_message(state, feed, data, height_after):
|
|
|
+ """PORT of onStreamMessage: skip heartbeats, ingest one row."""
|
|
|
+ if data == "ping":
|
|
|
+ return feed.scroll_top
|
|
|
+ ingest(state, {"id": int(data)})
|
|
|
+ return render(state, feed, mode="append", height_after=height_after)
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+# Behaviour tests
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+
|
|
|
+def ids(state):
|
|
|
+ return [evt["id"] for evt in state.events]
|
|
|
+
|
|
|
+
|
|
|
+class ChronologyTests(unittest.TestCase):
|
|
|
+
|
|
|
+ def test_initial_window_renders_oldest_first(self):
|
|
|
+ """The SSE replay / first JSON page arrives newest-first
|
|
|
+ ([30..1]); the feed must render [1..30] - oldest at the top."""
|
|
|
+ state = State()
|
|
|
+ for evt in server_page(30, 1, True)["events"]:
|
|
|
+ ingest(state, evt)
|
|
|
+ self.assertEqual(ids(state), list(range(1, 31)))
|
|
|
+ self.assertEqual(state.oldest, 1)
|
|
|
+
|
|
|
+ def test_live_appends_land_in_order(self):
|
|
|
+ state = State()
|
|
|
+ for i in range(30, 0, -1):
|
|
|
+ ingest(state, {"id": i})
|
|
|
+ ingest(state, {"id": 31})
|
|
|
+ ingest(state, {"id": 33})
|
|
|
+ ingest(state, {"id": 32}) # out of order arrival
|
|
|
+ self.assertEqual(ids(state)[-3:], [31, 32, 33])
|
|
|
+
|
|
|
+ def test_duplicate_ids_are_dropped(self):
|
|
|
+ state = State()
|
|
|
+ ingest(state, {"id": 5})
|
|
|
+ ingest(state, {"id": 5}) # EventSource reconnect replay
|
|
|
+ self.assertEqual(ids(state), [5])
|
|
|
+
|
|
|
+
|
|
|
+class ScrollStickTests(unittest.TestCase):
|
|
|
+
|
|
|
+ def test_first_render_with_rows_seeds_at_the_bottom(self):
|
|
|
+ state = State()
|
|
|
+ feed = Feed(scroll_height=0, client_height=600)
|
|
|
+ for evt in server_page(30, 1, True)["events"]:
|
|
|
+ ingest(state, evt)
|
|
|
+ render(state, feed, height_after=3000)
|
|
|
+ self.assertTrue(state.seeded)
|
|
|
+ self.assertEqual(feed.scroll_top, 3000)
|
|
|
+
|
|
|
+ def test_append_sticks_when_reading_the_tail(self):
|
|
|
+ state = State()
|
|
|
+ feed = Feed(scroll_height=3000, client_height=600)
|
|
|
+ state.seeded = True
|
|
|
+ feed.scroll_top = 2400 # exactly at the bottom
|
|
|
+ self.assertTrue(at_bottom(feed))
|
|
|
+ ingest(state, {"id": 31})
|
|
|
+ render(state, feed, mode="append", height_after=3100)
|
|
|
+ self.assertEqual(feed.scroll_top, 3100)
|
|
|
+
|
|
|
+ def test_append_within_threshold_still_sticks(self):
|
|
|
+ feed = Feed(scroll_height=3000, client_height=600, scroll_top=2360)
|
|
|
+ self.assertTrue(at_bottom(feed)) # 40px shy: close enough
|
|
|
+
|
|
|
+ def test_just_past_threshold_is_not_the_tail(self):
|
|
|
+ feed = Feed(scroll_height=3000, client_height=600, scroll_top=2351)
|
|
|
+ self.assertFalse(at_bottom(feed)) # 49px shy: reading history
|
|
|
+
|
|
|
+ def test_scrolled_up_reader_is_never_yanked(self):
|
|
|
+ state = State()
|
|
|
+ feed = Feed(scroll_height=3000, client_height=600, scroll_top=500)
|
|
|
+ state.seeded = True
|
|
|
+ self.assertFalse(at_bottom(feed))
|
|
|
+ ingest(state, {"id": 31})
|
|
|
+ render(state, feed, mode="append", height_after=3100)
|
|
|
+ self.assertEqual(feed.scroll_top, 500)
|
|
|
+
|
|
|
+ def test_short_feed_counts_as_at_bottom(self):
|
|
|
+ """A feed shorter than its viewport cannot be scrolled away
|
|
|
+ from the newest row."""
|
|
|
+ feed = Feed(scroll_height=100, client_height=600, scroll_top=0)
|
|
|
+ self.assertTrue(at_bottom(feed))
|
|
|
+
|
|
|
+ def test_heartbeat_does_not_touch_anything(self):
|
|
|
+ state = State()
|
|
|
+ feed = Feed(scroll_height=3000, client_height=600, scroll_top=2400)
|
|
|
+ state.seeded = True
|
|
|
+ before = (list(ids(state)), feed.scroll_top)
|
|
|
+ on_stream_message(state, feed, "ping", height_after=3100)
|
|
|
+ self.assertEqual((list(ids(state)), feed.scroll_top), before)
|
|
|
+
|
|
|
+
|
|
|
+class LoadMoreTests(unittest.TestCase):
|
|
|
+
|
|
|
+ def test_older_page_is_reversed_and_prepended(self):
|
|
|
+ """before=<oldest>=1 returns [60..31] newest-first on the
|
|
|
+ wire; the display must gain [31..60] above [1..30]."""
|
|
|
+ state = State()
|
|
|
+ feed = Feed(scroll_height=3000, client_height=600, scroll_top=500)
|
|
|
+ for evt in server_page(30, 1, True)["events"]:
|
|
|
+ ingest(state, evt)
|
|
|
+ state.seeded = True
|
|
|
+ render(state, feed, height_after=3000)
|
|
|
+ feed.scroll_top = 500 # the reader scrolled up into history
|
|
|
+ load_more(state, feed, server_page(60, 31, True), height_after=6000)
|
|
|
+ self.assertEqual(ids(state), list(range(1, 61)))
|
|
|
+ self.assertEqual(state.oldest, 1)
|
|
|
+ self.assertFalse(state.exhausted)
|
|
|
+
|
|
|
+ def test_prepend_keeps_the_viewport_anchored(self):
|
|
|
+ """Content added above must not move the rows under the
|
|
|
+ reader's cursor: scrollTop grows by the added height."""
|
|
|
+ state = State()
|
|
|
+ feed = Feed(scroll_height=3000, client_height=600, scroll_top=500)
|
|
|
+ state.seeded = True
|
|
|
+ self.assertFalse(at_bottom(feed))
|
|
|
+ delta = 3000
|
|
|
+ load_more(state, feed, server_page(60, 31, True),
|
|
|
+ height_after=3000 + delta)
|
|
|
+ self.assertEqual(feed.scroll_top, 500 + delta)
|
|
|
+
|
|
|
+ def test_prepend_while_at_bottom_stays_at_bottom(self):
|
|
|
+ state = State()
|
|
|
+ feed = Feed(scroll_height=3000, client_height=600, scroll_top=2400)
|
|
|
+ state.seeded = True
|
|
|
+ self.assertTrue(at_bottom(feed))
|
|
|
+ load_more(state, feed, server_page(60, 31, True), height_after=6000)
|
|
|
+ self.assertEqual(feed.scroll_top, 6000)
|
|
|
+
|
|
|
+ def test_exhausted_when_more_is_false(self):
|
|
|
+ state = State()
|
|
|
+ feed = Feed(scroll_height=3000, client_height=600, scroll_top=2400)
|
|
|
+ state.seeded = True
|
|
|
+ load_more(state, feed, server_page(60, 31, False), height_after=6000)
|
|
|
+ self.assertTrue(state.exhausted)
|
|
|
+
|
|
|
+ def test_empty_page_exhausts_too(self):
|
|
|
+ """A defensive guard: a page with no rows means no older
|
|
|
+ history, whatever the flag says."""
|
|
|
+ state = State()
|
|
|
+ feed = Feed(scroll_height=3000, client_height=600, scroll_top=2400)
|
|
|
+ state.seeded = True
|
|
|
+ ingest(state, {"id": 1})
|
|
|
+ load_more(state, feed, {"events": [], "more": True}, height_after=3000)
|
|
|
+ self.assertTrue(state.exhausted)
|
|
|
+
|
|
|
+
|
|
|
+class EndToEndSequenceTests(unittest.TestCase):
|
|
|
+ """The whole first-minute story: replay, live tail, history."""
|
|
|
+
|
|
|
+ def test_replay_then_append_then_history(self):
|
|
|
+ state = State()
|
|
|
+ feed = Feed(scroll_height=0, client_height=600)
|
|
|
+ # 1. SSE replay: most-recent-30 oldest-first on the wire.
|
|
|
+ for evt in server_page(30, 1, True)["events"]:
|
|
|
+ ingest(state, evt)
|
|
|
+ render(state, feed, height_after=3000)
|
|
|
+ self.assertEqual(ids(state), list(range(1, 31)))
|
|
|
+ self.assertEqual(feed.scroll_top, 3000) # seeded at the tail
|
|
|
+
|
|
|
+ # 2. Two live rows arrive; the reader is at the bottom.
|
|
|
+ on_stream_message(state, feed, "31", height_after=3100)
|
|
|
+ on_stream_message(state, feed, "32", height_after=3200)
|
|
|
+ self.assertEqual(ids(state)[-2:], [31, 32])
|
|
|
+ self.assertEqual(feed.scroll_top, 3200)
|
|
|
+
|
|
|
+ # 3. A heartbeat is ignored, then the reader scrolls up.
|
|
|
+ on_stream_message(state, feed, "ping", height_after=3200)
|
|
|
+ feed.scroll_top = 100
|
|
|
+ on_stream_message(state, feed, "33", height_after=3300)
|
|
|
+ self.assertEqual(feed.scroll_top, 100) # never yanked
|
|
|
+
|
|
|
+ # 4. Load older: before=1 -> [60..31] newest-first wire page.
|
|
|
+ self.assertEqual(state.oldest, 1)
|
|
|
+ load_more(state, feed, server_page(60, 31, False), height_after=6300)
|
|
|
+ self.assertEqual(ids(state), list(range(1, 61)))
|
|
|
+ self.assertTrue(state.exhausted)
|
|
|
+ self.assertEqual(feed.scroll_top, 100 + 3000) # anchored
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ unittest.main(verbosity=2)
|
saddle-Clanker-Saddle