Add a Share tab that hosts or joins a read-only live session through the rigdoctor-relay over WebSocket (QtWebSockets), gated by the Gitea access token. - gui/share_page.py: Start shared session (host: get a code, stream snapshot + health + inventory) and Enter share code (guest: view a host's data read-only) - core/share.py: host_full_frame / host_snapshot_frame + guest_html renderer - config: relay_url (default wss://rigdoctor.jesseyvanofferen.com) - setup: token now powers updates AND sharing — hint asks for read:user + read:repository scopes (relay validates the account via Gitea) - main_window: Share nav tab + socket cleanup on close - tests for the relay frame builders and guest HTML Verified end-to-end against the deployed relay (host code -> guest frame). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,46 @@
|
||||
"""Tests for M12 Tier 2 share server: token gating + endpoints."""
|
||||
|
||||
import json
|
||||
import threading
|
||||
import unittest
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
from rigdoctor.core import share
|
||||
|
||||
|
||||
class ShareServerTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.srv, self.token = share.make_server("127.0.0.1", 0)
|
||||
self.port = self.srv.server_address[1]
|
||||
self.thread = threading.Thread(target=self.srv.serve_forever, daemon=True)
|
||||
self.thread.start()
|
||||
|
||||
def tearDown(self):
|
||||
self.srv.shutdown()
|
||||
|
||||
def _url(self, path, token=None):
|
||||
q = f"?t={token}" if token else ""
|
||||
return f"http://127.0.0.1:{self.port}{path}{q}"
|
||||
|
||||
def test_requires_token(self):
|
||||
with self.assertRaises(urllib.error.HTTPError) as cm:
|
||||
urllib.request.urlopen(self._url("/api/snapshot"), timeout=10)
|
||||
self.assertEqual(cm.exception.code, 403)
|
||||
|
||||
def test_bad_token_rejected(self):
|
||||
with self.assertRaises(urllib.error.HTTPError) as cm:
|
||||
urllib.request.urlopen(self._url("/api/snapshot", "wrong"), timeout=10)
|
||||
self.assertEqual(cm.exception.code, 403)
|
||||
|
||||
def test_snapshot_with_token(self):
|
||||
data = json.load(urllib.request.urlopen(self._url("/api/snapshot", self.token), timeout=10))
|
||||
self.assertIn("groups", data)
|
||||
|
||||
def test_page_served(self):
|
||||
body = urllib.request.urlopen(self._url("/", self.token), timeout=10).read()
|
||||
self.assertIn(b"read-only share", body)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user